copy_export-20101213.diff
application/octet-stream
Filename: copy_export-20101213.diff
Type: application/octet-stream
Part: 0
Message:
Re: SQL/MED - file_fdw
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: unified
| File | + | − |
|---|---|---|
| src/backend/commands/copy.c | 0 | 0 |
| src/include/commands/copy.h | 0 | 0 |
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 7b8bee8..05bbce1 100644
*** a/src/backend/commands/copy.c
--- b/src/backend/commands/copy.c
*************** typedef struct CopyStateData
*** 142,150 ****
StringInfoData attribute_buf;
/* field raw data pointers found by COPY FROM */
!
! int max_fields;
! char ** raw_fields;
/*
* Similarly, line_buf holds the whole input line being processed. The
--- 142,149 ----
StringInfoData attribute_buf;
/* field raw data pointers found by COPY FROM */
! int max_fields;
! char **raw_fields;
/*
* Similarly, line_buf holds the whole input line being processed. The
*************** typedef struct CopyStateData
*** 167,175 ****
char *raw_buf;
int raw_buf_index; /* next byte to process */
int raw_buf_len; /* total # of bytes stored */
- } CopyStateData;
! typedef CopyStateData *CopyState;
/* DestReceiver for COPY (SELECT) TO */
typedef struct
--- 166,186 ----
char *raw_buf;
int raw_buf_index; /* next byte to process */
int raw_buf_len; /* total # of bytes stored */
! /*
! * The definition of input functions and default expressions are stored
! * in these variables.
! */
! EState *estate;
! AttrNumber num_defaults;
! bool file_has_oids;
! FmgrInfo oid_in_function;
! Oid oid_typioparam;
! FmgrInfo *in_functions;
! Oid *typioparams;
! int *defmap;
! ExprState **defexprs; /* array of default att expressions */
! } CopyStateData;
/* DestReceiver for COPY (SELECT) TO */
typedef struct
*************** static const char BinarySignature[11] =
*** 248,253 ****
--- 259,270 ----
/* non-export function prototypes */
+ static CopyState BeginCopy(bool is_from,
+ Relation rel, Node *raw_query, const char *queryString,
+ const char *filename, List *attnamelist, List *options);
+ static CopyState BeginCopyTo(Relation rel, Node *query, const char *queryString,
+ const char *filename, List *attnamelist, List *options);
+ static void EndCopyTo(CopyState cstate);
static void DoCopyTo(CopyState cstate);
static void CopyTo(CopyState cstate);
static void CopyOneRowTo(CopyState cstate, Oid tupleOid,
*************** CopyLoadRawBuf(CopyState cstate)
*** 718,730 ****
* Do not allow the copy if user doesn't have proper permission to access
* the table or the specifically requested columns.
*/
! uint64
! DoCopy(const CopyStmt *stmt, const char *queryString)
{
CopyState cstate;
- bool is_from = stmt->is_from;
- bool pipe = (stmt->filename == NULL);
- List *attnamelist = stmt->attlist;
List *force_quote = NIL;
List *force_notnull = NIL;
bool force_quote_all = false;
--- 735,746 ----
* Do not allow the copy if user doesn't have proper permission to access
* the table or the specifically requested columns.
*/
! static CopyState
! BeginCopy(bool is_from,
! Relation rel, Node *raw_query, const char *queryString,
! const char *filename, List *attnamelist, List *options)
{
CopyState cstate;
List *force_quote = NIL;
List *force_notnull = NIL;
bool force_quote_all = false;
*************** DoCopy(const CopyStmt *stmt, const char
*** 733,745 ****
ListCell *option;
TupleDesc tupDesc;
int num_phys_attrs;
- uint64 processed;
/* Allocate workspace and zero all fields */
cstate = (CopyStateData *) palloc0(sizeof(CopyStateData));
/* Extract options from the statement node tree */
! foreach(option, stmt->options)
{
DefElem *defel = (DefElem *) lfirst(option);
--- 749,760 ----
ListCell *option;
TupleDesc tupDesc;
int num_phys_attrs;
/* Allocate workspace and zero all fields */
cstate = (CopyStateData *) palloc0(sizeof(CopyStateData));
/* Extract options from the statement node tree */
! foreach(option, options)
{
DefElem *defel = (DefElem *) lfirst(option);
*************** DoCopy(const CopyStmt *stmt, const char
*** 980,1005 ****
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("CSV quote character must not appear in the NULL specification")));
! /* Disallow file COPY except to superusers. */
! if (!pipe && !superuser())
! ereport(ERROR,
! (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
! errmsg("must be superuser to COPY to or from a file"),
! errhint("Anyone can COPY to stdout or from stdin. "
! "psql's \\copy command also works for anyone.")));
!
! if (stmt->relation)
{
RangeTblEntry *rte;
List *attnums;
ListCell *cur;
! Assert(!stmt->query);
! cstate->queryDesc = NULL;
! /* Open and lock the relation, using the appropriate lock type. */
! cstate->rel = heap_openrv(stmt->relation,
! (is_from ? RowExclusiveLock : AccessShareLock));
tupDesc = RelationGetDescr(cstate->rel);
--- 995,1009 ----
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("CSV quote character must not appear in the NULL specification")));
! if (rel)
{
RangeTblEntry *rte;
List *attnums;
ListCell *cur;
! Assert(!raw_query);
! cstate->rel = rel;
tupDesc = RelationGetDescr(cstate->rel);
*************** DoCopy(const CopyStmt *stmt, const char
*** 1058,1064 ****
* function and is executed repeatedly. (See also the same hack in
* DECLARE CURSOR and PREPARE.) XXX FIXME someday.
*/
! rewritten = pg_analyze_and_rewrite((Node *) copyObject(stmt->query),
queryString, NULL, 0);
/* We don't expect more or less than one result query */
--- 1062,1068 ----
* function and is executed repeatedly. (See also the same hack in
* DECLARE CURSOR and PREPARE.) XXX FIXME someday.
*/
! rewritten = pg_analyze_and_rewrite((Node *) copyObject(raw_query),
queryString, NULL, 0);
/* We don't expect more or less than one result query */
*************** DoCopy(const CopyStmt *stmt, const char
*** 1160,1171 ****
}
}
- /* Set up variables to avoid per-attribute overhead. */
- initStringInfo(&cstate->attribute_buf);
- initStringInfo(&cstate->line_buf);
- cstate->line_buf_converted = false;
- cstate->raw_buf = (char *) palloc(RAW_BUF_SIZE + 1);
- cstate->raw_buf_index = cstate->raw_buf_len = 0;
cstate->processed = 0;
/*
--- 1164,1169 ----
*************** DoCopy(const CopyStmt *stmt, const char
*** 1181,1202 ****
cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->client_encoding);
cstate->copy_dest = COPY_FILE; /* default */
! cstate->filename = stmt->filename;
if (is_from)
CopyFrom(cstate); /* copy from file to database */
else
DoCopyTo(cstate); /* copy from database to file */
/*
! * Close the relation or query. If reading, we can release the
! * AccessShareLock we got; if writing, we should hold the lock until end
! * of transaction to ensure that updates will be committed before lock is
! * released.
*/
! if (cstate->rel)
! heap_close(cstate->rel, (is_from ? NoLock : AccessShareLock));
! else
{
/* Close down the query and free resources. */
ExecutorEnd(cstate->queryDesc);
--- 1179,1681 ----
cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->client_encoding);
cstate->copy_dest = COPY_FILE; /* default */
! cstate->filename = (filename ? pstrdup(filename) : NULL);
!
! return cstate;
! }
!
! CopyState
! BeginCopyFrom(Relation rel, const char *filename, List *attnamelist, List *options)
! {
! CopyState cstate;
! bool pipe = (filename == NULL);
! TupleDesc tupDesc;
! int num_phys_attrs;
! Form_pg_attribute *attr;
! AttrNumber attr_count;
! int attnum;
! EState *estate;
! Oid in_func_oid;
!
! cstate = BeginCopy(true, rel, NULL, NULL, filename, attnamelist, options);
!
! /* Initialize state variables */
! cstate->fe_eof = false;
! cstate->eol_type = EOL_UNKNOWN;
! cstate->cur_relname = RelationGetRelationName(cstate->rel);
! cstate->cur_lineno = 0;
! cstate->cur_attname = NULL;
! cstate->cur_attval = NULL;
!
! /* Set up variables to avoid per-attribute overhead. */
! initStringInfo(&cstate->attribute_buf);
! initStringInfo(&cstate->line_buf);
! cstate->line_buf_converted = false;
! cstate->raw_buf = (char *) palloc(RAW_BUF_SIZE + 1);
! cstate->raw_buf_index = cstate->raw_buf_len = 0;
!
! tupDesc = RelationGetDescr(cstate->rel);
! attr = tupDesc->attrs;
! num_phys_attrs = tupDesc->natts;
! attr_count = list_length(cstate->attnumlist);
! cstate->num_defaults = 0;
!
! /*
! * Pick up the required catalog information for each attribute in the
! * relation, including the input function, the element type (to pass to
! * the input function), and info about defaults and constraints. (Which
! * input function we use depends on text/binary format choice.)
! */
! cstate->in_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
! cstate->typioparams = (Oid *) palloc(num_phys_attrs * sizeof(Oid));
! cstate->defmap = (int *) palloc(num_phys_attrs * sizeof(int));
! cstate->defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));
!
! /* We need a ResultRelInfo to check constraints. */
! estate = cstate->estate = CreateExecutorState();
!
! for (attnum = 1; attnum <= num_phys_attrs; attnum++)
! {
! /* We don't need info for dropped attributes */
! if (attr[attnum - 1]->attisdropped)
! continue;
!
! /* Fetch the input function and typioparam info */
! if (cstate->binary)
! getTypeBinaryInputInfo(attr[attnum - 1]->atttypid,
! &in_func_oid, &cstate->typioparams[attnum - 1]);
! else
! getTypeInputInfo(attr[attnum - 1]->atttypid,
! &in_func_oid, &cstate->typioparams[attnum - 1]);
! fmgr_info(in_func_oid, &cstate->in_functions[attnum - 1]);
!
! /* Get default info if needed */
! if (!list_member_int(cstate->attnumlist, attnum))
! {
! /* attribute is NOT to be copied from input */
! /* use default value if one exists */
! Node *defexpr = build_column_default(cstate->rel, attnum);
!
! if (defexpr != NULL)
! {
! cstate->defexprs[cstate->num_defaults] =
! ExecPrepareExpr((Expr *) defexpr, estate);
! cstate->defmap[cstate->num_defaults] = attnum - 1;
! cstate->num_defaults++;
! }
! }
! }
!
! if (pipe)
! {
! if (whereToSendOutput == DestRemote)
! ReceiveCopyBegin(cstate);
! else
! cstate->copy_file = stdin;
! }
! else
! {
! struct stat st;
!
! cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_R);
!
! if (cstate->copy_file == NULL)
! ereport(ERROR,
! (errcode_for_file_access(),
! errmsg("could not open file \"%s\" for reading: %m",
! cstate->filename)));
!
! fstat(fileno(cstate->copy_file), &st);
! if (S_ISDIR(st.st_mode))
! ereport(ERROR,
! (errcode(ERRCODE_WRONG_OBJECT_TYPE),
! errmsg("\"%s\" is a directory", cstate->filename)));
! }
!
! if (!cstate->binary)
! cstate->file_has_oids = cstate->oids; /* must rely on user to tell us... */
! else
! {
! /* Read and verify binary header */
! char readSig[11];
! int32 tmp;
!
! /* Signature */
! if (CopyGetData(cstate, readSig, 11, 11) != 11 ||
! memcmp(readSig, BinarySignature, 11) != 0)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("COPY file signature not recognized")));
! /* Flags field */
! if (!CopyGetInt32(cstate, &tmp))
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid COPY file header (missing flags)")));
! cstate->file_has_oids = (tmp & (1 << 16)) != 0;
! tmp &= ~(1 << 16);
! if ((tmp >> 16) != 0)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("unrecognized critical flags in COPY file header")));
! /* Header extension length */
! if (!CopyGetInt32(cstate, &tmp) ||
! tmp < 0)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid COPY file header (missing length)")));
! /* Skip extension header, if present */
! while (tmp-- > 0)
! {
! if (CopyGetData(cstate, readSig, 1, 1) != 1)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid COPY file header (wrong length)")));
! }
! }
!
! if (cstate->file_has_oids && cstate->binary)
! {
! Oid in_func_oid;
! getTypeBinaryInputInfo(OIDOID,
! &in_func_oid, &cstate->oid_typioparam);
! fmgr_info(in_func_oid, &cstate->oid_in_function);
! }
!
! /* create workspace for CopyReadAttributes results */
! if (!cstate->binary)
! {
! int nfields = cstate->file_has_oids ? (attr_count + 1) : attr_count;
!
! cstate->max_fields = nfields;
! cstate->raw_fields = (char **) palloc(nfields * sizeof(char *));
! }
!
! return cstate;
! }
!
! static CopyState
! BeginCopyTo(Relation rel, Node *query, const char *queryString,
! const char *filename, List *attnamelist, List *options)
! {
! return BeginCopy(false, rel, query, queryString,
! filename, attnamelist, options);
! }
!
! /* return false if no more tuples */
! bool
! NextCopyFrom(CopyState cstate, Datum *values, bool *nulls, Oid *oid)
! {
! TupleDesc tupDesc;
! Form_pg_attribute *attr;
! AttrNumber num_phys_attrs,
! attr_count,
! num_defaults = cstate->num_defaults;
! FmgrInfo *in_functions = cstate->in_functions;
! Oid *typioparams = cstate->typioparams;
! int i;
! int nfields;
! char **field_strings;
! bool isnull;
! int *defmap = cstate->defmap;
! ExprState **defexprs = cstate->defexprs;
! ExprContext *econtext; /* used for ExecEvalExpr for default atts */
!
! /* on input just throw the header line away */
! if (cstate->cur_lineno == 0 && cstate->header_line)
! {
! cstate->cur_lineno++;
! if (CopyReadLine(cstate))
! return false; /* done */
! }
!
! cstate->cur_lineno++;
!
! tupDesc = RelationGetDescr(cstate->rel);
! attr = tupDesc->attrs;
! num_phys_attrs = tupDesc->natts;
! attr_count = list_length(cstate->attnumlist);
! nfields = cstate->file_has_oids ? (attr_count + 1) : attr_count;
!
! /* Initialize all values for row to NULL */
! MemSet(values, 0, num_phys_attrs * sizeof(Datum));
! MemSet(nulls, true, num_phys_attrs * sizeof(bool));
!
! if (!cstate->binary)
! {
! ListCell *cur;
! int fldct;
! int fieldno;
! char *string;
!
! /*
! * Actually read the line into memory here.
! * EOF at start of line means we're done. If we see EOF after
! * some characters, we act as though it was newline followed by
! * EOF, ie, process the line and then exit loop on next iteration.
! */
! if (CopyReadLine(cstate) && cstate->line_buf.len == 0)
! return false;
!
! /* Parse the line into de-escaped field values */
! if (cstate->csv_mode)
! fldct = CopyReadAttributesCSV(cstate);
! else
! fldct = CopyReadAttributesText(cstate);
!
! /* check for overflowing fields */
! if (nfields > 0 && fldct > nfields)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("extra data after last expected column")));
!
! fieldno = 0;
! field_strings = cstate->raw_fields;
!
! /* Read the OID field if present */
! if (cstate->file_has_oids)
! {
! if (fieldno >= fldct)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("missing data for OID column")));
! string = field_strings[fieldno++];
!
! if (string == NULL)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("null OID in COPY data")));
! else
! {
! cstate->cur_attname = "oid";
! cstate->cur_attval = string;
! *oid = DatumGetObjectId(DirectFunctionCall1(oidin,
! CStringGetDatum(string)));
! if (*oid == InvalidOid)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid OID in COPY data")));
! cstate->cur_attname = NULL;
! cstate->cur_attval = NULL;
! }
! }
!
! /* Loop to read the user attributes on the line. */
! foreach(cur, cstate->attnumlist)
! {
! int attnum = lfirst_int(cur);
! int m = attnum - 1;
!
! if (fieldno >= fldct)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("missing data for column \"%s\"",
! NameStr(attr[m]->attname))));
! string = field_strings[fieldno++];
!
! if (cstate->csv_mode && string == NULL &&
! cstate->force_notnull_flags[m])
! {
! /* Go ahead and read the NULL string */
! string = cstate->null_print;
! }
!
! cstate->cur_attname = NameStr(attr[m]->attname);
! cstate->cur_attval = string;
! values[m] = InputFunctionCall(&in_functions[m],
! string,
! typioparams[m],
! attr[m]->atttypmod);
! if (string != NULL)
! nulls[m] = false;
! cstate->cur_attname = NULL;
! cstate->cur_attval = NULL;
! }
!
! Assert(fieldno == nfields);
! }
! else
! {
! /* binary */
! int16 fld_count;
! ListCell *cur;
!
! if (!CopyGetInt16(cstate, &fld_count))
! {
! /* EOF detected (end of file, or protocol-level EOF) */
! return false;
! }
!
! if (fld_count == -1)
! {
! /*
! * Received EOF marker. In a V3-protocol copy, wait for
! * the protocol-level EOF, and complain if it doesn't come
! * immediately. This ensures that we correctly handle
! * CopyFail, if client chooses to send that now.
! *
! * Note that we MUST NOT try to read more data in an
! * old-protocol copy, since there is no protocol-level EOF
! * marker then. We could go either way for copy from file,
! * but choose to throw error if there's data after the EOF
! * marker, for consistency with the new-protocol case.
! */
! char dummy;
!
! if (cstate->copy_dest != COPY_OLD_FE &&
! CopyGetData(cstate, &dummy, 1, 1) > 0)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("received copy data after EOF marker")));
! return false;
! }
!
! if (fld_count != attr_count)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("row field count is %d, expected %d",
! (int) fld_count, attr_count)));
!
! if (cstate->file_has_oids)
! {
! cstate->cur_attname = "oid";
! *oid = DatumGetObjectId(CopyReadBinaryAttribute(cstate,
! 0,
! &cstate->oid_in_function,
! cstate->oid_typioparam,
! -1,
! &isnull));
! if (isnull || *oid == InvalidOid)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid OID in COPY data")));
! cstate->cur_attname = NULL;
! }
!
! i = 0;
! foreach(cur, cstate->attnumlist)
! {
! int attnum = lfirst_int(cur);
! int m = attnum - 1;
!
! cstate->cur_attname = NameStr(attr[m]->attname);
! i++;
! values[m] = CopyReadBinaryAttribute(cstate,
! i,
! &in_functions[m],
! typioparams[m],
! attr[m]->atttypmod,
! &nulls[m]);
! cstate->cur_attname = NULL;
! }
! }
!
! /*
! * Now compute and insert any defaults available for the columns not
! * provided by the input data. Anything not processed here or above
! * will remain NULL.
! */
! econtext = GetPerTupleExprContext(cstate->estate);
! for (i = 0; i < num_defaults; i++)
! {
! values[defmap[i]] = ExecEvalExpr(defexprs[i], econtext,
! &nulls[defmap[i]], NULL);
! }
!
! return true;
! }
!
! uint64
! DoCopy(const CopyStmt *stmt, const char *queryString)
! {
! CopyState cstate;
! bool is_from = stmt->is_from;
! bool pipe = (stmt->filename == NULL);
! Relation rel;
! uint64 processed;
!
! /* Disallow file COPY except to superusers. */
! if (!pipe && !superuser())
! ereport(ERROR,
! (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
! errmsg("must be superuser to COPY to or from a file"),
! errhint("Anyone can COPY to stdout or from stdin. "
! "psql's \\copy command also works for anyone.")));
!
! if (stmt->relation)
! {
! Assert(!stmt->query);
!
! /* Open and lock the relation, using the appropriate lock type. */
! rel = heap_openrv(stmt->relation,
! (is_from ? RowExclusiveLock : AccessShareLock));
! }
! else
! {
! Assert(stmt->query);
!
! rel = NULL;
! }
if (is_from)
+ {
+ cstate = BeginCopyFrom(rel, stmt->filename, stmt->attlist, stmt->options);
CopyFrom(cstate); /* copy from file to database */
+ processed = cstate->processed;
+ EndCopyFrom(cstate);
+ }
else
+ {
+ cstate = BeginCopyTo(rel, stmt->query, queryString, stmt->filename, stmt->attlist, stmt->options);
DoCopyTo(cstate); /* copy from database to file */
+ processed = cstate->processed;
+ EndCopyTo(cstate);
+ }
/*
! * Close the relation. If reading, we can release the AccessShareLock we got;
! * if writing, we should hold the lock until end of transaction to ensure that
! * updates will be committed before lock is released.
*/
! if (rel != NULL)
! heap_close(rel, (is_from ? NoLock : AccessShareLock));
!
! return processed;
! }
!
! void
! EndCopyFrom(CopyState cstate)
! {
! FreeExecutorState(cstate->estate);
!
! /* Clean up storage */
! if (cstate->filename)
! {
! if (FreeFile(cstate->copy_file))
! ereport(ERROR,
! (errcode_for_file_access(),
! errmsg("could not read from file \"%s\": %m",
! cstate->filename)));
! pfree(cstate->filename);
! }
! if (!cstate->binary)
! pfree(cstate->raw_fields);
! pfree(cstate->attribute_buf.data);
! pfree(cstate->line_buf.data);
! pfree(cstate->raw_buf);
! pfree(cstate->in_functions);
! pfree(cstate->typioparams);
! pfree(cstate->defmap);
! pfree(cstate->defexprs);
! pfree(cstate);
! }
!
! static void
! EndCopyTo(CopyState cstate)
! {
! /*
! * Close the relation or query. We can release the AccessShareLock we got.
! */
! if (cstate->rel == NULL)
{
/* Close down the query and free resources. */
ExecutorEnd(cstate->queryDesc);
*************** DoCopy(const CopyStmt *stmt, const char
*** 1204,1221 ****
PopActiveSnapshot();
}
! /* Clean up storage (probably not really necessary) */
! processed = cstate->processed;
!
! pfree(cstate->attribute_buf.data);
! pfree(cstate->line_buf.data);
! pfree(cstate->raw_buf);
pfree(cstate);
-
- return processed;
}
-
/*
* This intermediate routine exists mainly to localize the effects of setjmp
* so we don't need to plaster a lot of variables with "volatile".
--- 1683,1694 ----
PopActiveSnapshot();
}
! /* Clean up storage */
! if (cstate->filename)
! pfree(cstate->filename);
pfree(cstate);
}
/*
* This intermediate routine exists mainly to localize the effects of setjmp
* so we don't need to plaster a lot of variables with "volatile".
*************** limit_printout_length(const char *str)
*** 1666,1698 ****
static void
CopyFrom(CopyState cstate)
{
- bool pipe = (cstate->filename == NULL);
HeapTuple tuple;
TupleDesc tupDesc;
- Form_pg_attribute *attr;
- AttrNumber num_phys_attrs,
- attr_count,
- num_defaults;
- FmgrInfo *in_functions;
- FmgrInfo oid_in_function;
- Oid *typioparams;
- Oid oid_typioparam;
- int attnum;
- int i;
- Oid in_func_oid;
Datum *values;
bool *nulls;
- int nfields;
- char **field_strings;
bool done = false;
- bool isnull;
ResultRelInfo *resultRelInfo;
! EState *estate = CreateExecutorState(); /* for ExecConstraints() */
TupleTableSlot *slot;
- bool file_has_oids;
- int *defmap;
- ExprState **defexprs; /* array of default att expressions */
- ExprContext *econtext; /* used for ExecEvalExpr for default atts */
MemoryContext oldcontext = CurrentMemoryContext;
ErrorContextCallback errcontext;
CommandId mycid = GetCurrentCommandId(true);
--- 2139,2152 ----
static void
CopyFrom(CopyState cstate)
{
HeapTuple tuple;
TupleDesc tupDesc;
Datum *values;
bool *nulls;
bool done = false;
ResultRelInfo *resultRelInfo;
! EState *estate = cstate->estate; /* for ExecConstraints() */
TupleTableSlot *slot;
MemoryContext oldcontext = CurrentMemoryContext;
ErrorContextCallback errcontext;
CommandId mycid = GetCurrentCommandId(true);
*************** CopyFrom(CopyState cstate)
*** 1720,1725 ****
--- 2174,2181 ----
RelationGetRelationName(cstate->rel))));
}
+ tupDesc = RelationGetDescr(cstate->rel);
+
/*----------
* Check to see if we can avoid writing WAL
*
*************** CopyFrom(CopyState cstate)
*** 1755,1792 ****
hi_options |= HEAP_INSERT_SKIP_WAL;
}
- if (pipe)
- {
- if (whereToSendOutput == DestRemote)
- ReceiveCopyBegin(cstate);
- else
- cstate->copy_file = stdin;
- }
- else
- {
- struct stat st;
-
- cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_R);
-
- if (cstate->copy_file == NULL)
- ereport(ERROR,
- (errcode_for_file_access(),
- errmsg("could not open file \"%s\" for reading: %m",
- cstate->filename)));
-
- fstat(fileno(cstate->copy_file), &st);
- if (S_ISDIR(st.st_mode))
- ereport(ERROR,
- (errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("\"%s\" is a directory", cstate->filename)));
- }
-
- tupDesc = RelationGetDescr(cstate->rel);
- attr = tupDesc->attrs;
- num_phys_attrs = tupDesc->natts;
- attr_count = list_length(cstate->attnumlist);
- num_defaults = 0;
-
/*
* We need a ResultRelInfo so we can use the regular executor's
* index-entry-making machinery. (There used to be a huge amount of code
--- 2211,2216 ----
*************** CopyFrom(CopyState cstate)
*** 1815,1865 ****
slot = ExecInitExtraTupleSlot(estate);
ExecSetSlotDescriptor(slot, tupDesc);
- econtext = GetPerTupleExprContext(estate);
-
- /*
- * Pick up the required catalog information for each attribute in the
- * relation, including the input function, the element type (to pass to
- * the input function), and info about defaults and constraints. (Which
- * input function we use depends on text/binary format choice.)
- */
- in_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
- typioparams = (Oid *) palloc(num_phys_attrs * sizeof(Oid));
- defmap = (int *) palloc(num_phys_attrs * sizeof(int));
- defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));
-
- for (attnum = 1; attnum <= num_phys_attrs; attnum++)
- {
- /* We don't need info for dropped attributes */
- if (attr[attnum - 1]->attisdropped)
- continue;
-
- /* Fetch the input function and typioparam info */
- if (cstate->binary)
- getTypeBinaryInputInfo(attr[attnum - 1]->atttypid,
- &in_func_oid, &typioparams[attnum - 1]);
- else
- getTypeInputInfo(attr[attnum - 1]->atttypid,
- &in_func_oid, &typioparams[attnum - 1]);
- fmgr_info(in_func_oid, &in_functions[attnum - 1]);
-
- /* Get default info if needed */
- if (!list_member_int(cstate->attnumlist, attnum))
- {
- /* attribute is NOT to be copied from input */
- /* use default value if one exists */
- Node *defexpr = build_column_default(cstate->rel, attnum);
-
- if (defexpr != NULL)
- {
- defexprs[num_defaults] = ExecPrepareExpr((Expr *) defexpr,
- estate);
- defmap[num_defaults] = attnum - 1;
- num_defaults++;
- }
- }
- }
-
/* Prepare to catch AFTER triggers. */
AfterTriggerBeginQuery();
--- 2239,2244 ----
*************** CopyFrom(CopyState cstate)
*** 1871,1942 ****
*/
ExecBSInsertTriggers(estate, resultRelInfo);
! if (!cstate->binary)
! file_has_oids = cstate->oids; /* must rely on user to tell us... */
! else
! {
! /* Read and verify binary header */
! char readSig[11];
! int32 tmp;
!
! /* Signature */
! if (CopyGetData(cstate, readSig, 11, 11) != 11 ||
! memcmp(readSig, BinarySignature, 11) != 0)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("COPY file signature not recognized")));
! /* Flags field */
! if (!CopyGetInt32(cstate, &tmp))
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid COPY file header (missing flags)")));
! file_has_oids = (tmp & (1 << 16)) != 0;
! tmp &= ~(1 << 16);
! if ((tmp >> 16) != 0)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("unrecognized critical flags in COPY file header")));
! /* Header extension length */
! if (!CopyGetInt32(cstate, &tmp) ||
! tmp < 0)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid COPY file header (missing length)")));
! /* Skip extension header, if present */
! while (tmp-- > 0)
! {
! if (CopyGetData(cstate, readSig, 1, 1) != 1)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid COPY file header (wrong length)")));
! }
! }
!
! if (file_has_oids && cstate->binary)
! {
! getTypeBinaryInputInfo(OIDOID,
! &in_func_oid, &oid_typioparam);
! fmgr_info(in_func_oid, &oid_in_function);
! }
!
! values = (Datum *) palloc(num_phys_attrs * sizeof(Datum));
! nulls = (bool *) palloc(num_phys_attrs * sizeof(bool));
!
! /* create workspace for CopyReadAttributes results */
! nfields = file_has_oids ? (attr_count + 1) : attr_count;
! if (! cstate->binary)
! {
! cstate->max_fields = nfields;
! cstate->raw_fields = (char **) palloc(nfields * sizeof(char *));
! }
!
! /* Initialize state variables */
! cstate->fe_eof = false;
! cstate->eol_type = EOL_UNKNOWN;
! cstate->cur_relname = RelationGetRelationName(cstate->rel);
! cstate->cur_lineno = 0;
! cstate->cur_attname = NULL;
! cstate->cur_attval = NULL;
bistate = GetBulkInsertState();
--- 2250,2257 ----
*/
ExecBSInsertTriggers(estate, resultRelInfo);
! values = (Datum *) palloc(tupDesc->natts * sizeof(Datum));
! nulls = (bool *) palloc(tupDesc->natts * sizeof(bool));
bistate = GetBulkInsertState();
*************** CopyFrom(CopyState cstate)
*** 1946,1958 ****
errcontext.previous = error_context_stack;
error_context_stack = &errcontext;
- /* on input just throw the header line away */
- if (cstate->header_line)
- {
- cstate->cur_lineno++;
- done = CopyReadLine(cstate);
- }
-
while (!done)
{
bool skip_tuple;
--- 2261,2266 ----
*************** CopyFrom(CopyState cstate)
*** 1960,2166 ****
CHECK_FOR_INTERRUPTS();
- cstate->cur_lineno++;
-
/* Reset the per-tuple exprcontext */
ResetPerTupleExprContext(estate);
/* Switch into its memory context */
MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
! /* Initialize all values for row to NULL */
! MemSet(values, 0, num_phys_attrs * sizeof(Datum));
! MemSet(nulls, true, num_phys_attrs * sizeof(bool));
!
! if (!cstate->binary)
! {
! ListCell *cur;
! int fldct;
! int fieldno;
! char *string;
!
! /* Actually read the line into memory here */
! done = CopyReadLine(cstate);
!
! /*
! * EOF at start of line means we're done. If we see EOF after
! * some characters, we act as though it was newline followed by
! * EOF, ie, process the line and then exit loop on next iteration.
! */
! if (done && cstate->line_buf.len == 0)
! break;
!
! /* Parse the line into de-escaped field values */
! if (cstate->csv_mode)
! fldct = CopyReadAttributesCSV(cstate);
! else
! fldct = CopyReadAttributesText(cstate);
!
! /* check for overflowing fields */
! if (nfields > 0 && fldct > nfields)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("extra data after last expected column")));
!
! fieldno = 0;
! field_strings = cstate->raw_fields;
!
! /* Read the OID field if present */
! if (file_has_oids)
! {
! if (fieldno >= fldct)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("missing data for OID column")));
! string = field_strings[fieldno++];
!
! if (string == NULL)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("null OID in COPY data")));
! else
! {
! cstate->cur_attname = "oid";
! cstate->cur_attval = string;
! loaded_oid = DatumGetObjectId(DirectFunctionCall1(oidin,
! CStringGetDatum(string)));
! if (loaded_oid == InvalidOid)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid OID in COPY data")));
! cstate->cur_attname = NULL;
! cstate->cur_attval = NULL;
! }
! }
!
! /* Loop to read the user attributes on the line. */
! foreach(cur, cstate->attnumlist)
! {
! int attnum = lfirst_int(cur);
! int m = attnum - 1;
!
! if (fieldno >= fldct)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("missing data for column \"%s\"",
! NameStr(attr[m]->attname))));
! string = field_strings[fieldno++];
!
! if (cstate->csv_mode && string == NULL &&
! cstate->force_notnull_flags[m])
! {
! /* Go ahead and read the NULL string */
! string = cstate->null_print;
! }
!
! cstate->cur_attname = NameStr(attr[m]->attname);
! cstate->cur_attval = string;
! values[m] = InputFunctionCall(&in_functions[m],
! string,
! typioparams[m],
! attr[m]->atttypmod);
! if (string != NULL)
! nulls[m] = false;
! cstate->cur_attname = NULL;
! cstate->cur_attval = NULL;
! }
!
! Assert(fieldno == nfields);
! }
! else
! {
! /* binary */
! int16 fld_count;
! ListCell *cur;
!
! if (!CopyGetInt16(cstate, &fld_count))
! {
! /* EOF detected (end of file, or protocol-level EOF) */
! done = true;
! break;
! }
!
! if (fld_count == -1)
! {
! /*
! * Received EOF marker. In a V3-protocol copy, wait for
! * the protocol-level EOF, and complain if it doesn't come
! * immediately. This ensures that we correctly handle
! * CopyFail, if client chooses to send that now.
! *
! * Note that we MUST NOT try to read more data in an
! * old-protocol copy, since there is no protocol-level EOF
! * marker then. We could go either way for copy from file,
! * but choose to throw error if there's data after the EOF
! * marker, for consistency with the new-protocol case.
! */
! char dummy;
!
! if (cstate->copy_dest != COPY_OLD_FE &&
! CopyGetData(cstate, &dummy, 1, 1) > 0)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("received copy data after EOF marker")));
! done = true;
! break;
! }
!
! if (fld_count != attr_count)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("row field count is %d, expected %d",
! (int) fld_count, attr_count)));
!
! if (file_has_oids)
! {
! cstate->cur_attname = "oid";
! loaded_oid =
! DatumGetObjectId(CopyReadBinaryAttribute(cstate,
! 0,
! &oid_in_function,
! oid_typioparam,
! -1,
! &isnull));
! if (isnull || loaded_oid == InvalidOid)
! ereport(ERROR,
! (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
! errmsg("invalid OID in COPY data")));
! cstate->cur_attname = NULL;
! }
!
! i = 0;
! foreach(cur, cstate->attnumlist)
! {
! int attnum = lfirst_int(cur);
! int m = attnum - 1;
!
! cstate->cur_attname = NameStr(attr[m]->attname);
! i++;
! values[m] = CopyReadBinaryAttribute(cstate,
! i,
! &in_functions[m],
! typioparams[m],
! attr[m]->atttypmod,
! &nulls[m]);
! cstate->cur_attname = NULL;
! }
! }
!
! /*
! * Now compute and insert any defaults available for the columns not
! * provided by the input data. Anything not processed here or above
! * will remain NULL.
! */
! for (i = 0; i < num_defaults; i++)
! {
! values[defmap[i]] = ExecEvalExpr(defexprs[i], econtext,
! &nulls[defmap[i]], NULL);
! }
/* And now we can form the input tuple. */
tuple = heap_form_tuple(tupDesc, values, nulls);
! if (cstate->oids && file_has_oids)
HeapTupleSetOid(tuple, loaded_oid);
/* Triggers and stuff need to be invoked in query context. */
--- 2268,2287 ----
CHECK_FOR_INTERRUPTS();
/* Reset the per-tuple exprcontext */
ResetPerTupleExprContext(estate);
/* Switch into its memory context */
MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
! done = !NextCopyFrom(cstate, values, nulls, &loaded_oid);
! if (done)
! break;
/* And now we can form the input tuple. */
tuple = heap_form_tuple(tupDesc, values, nulls);
! if (cstate->oids && cstate->file_has_oids)
HeapTupleSetOid(tuple, loaded_oid);
/* Triggers and stuff need to be invoked in query context. */
*************** CopyFrom(CopyState cstate)
*** 2233,2261 ****
pfree(values);
pfree(nulls);
- if (! cstate->binary)
- pfree(cstate->raw_fields);
-
- pfree(in_functions);
- pfree(typioparams);
- pfree(defmap);
- pfree(defexprs);
ExecResetTupleTable(estate->es_tupleTable, false);
ExecCloseIndices(resultRelInfo);
- FreeExecutorState(estate);
-
- if (!pipe)
- {
- if (FreeFile(cstate->copy_file))
- ereport(ERROR,
- (errcode_for_file_access(),
- errmsg("could not read from file \"%s\": %m",
- cstate->filename)));
- }
-
/*
* If we skipped writing WAL, then we need to sync the heap (but not
* indexes since those use WAL anyway)
--- 2354,2364 ----
diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h
index 6d409e8..4d9bd60 100644
*** a/src/include/commands/copy.h
--- b/src/include/commands/copy.h
***************
*** 18,25 ****
--- 18,32 ----
#include "tcop/dest.h"
+ typedef struct CopyStateData *CopyState;
+
extern uint64 DoCopy(const CopyStmt *stmt, const char *queryString);
+ extern CopyState BeginCopyFrom(Relation rel,
+ const char *filename, List *attnamelist, List *options);
+ extern void EndCopyFrom(CopyState cstate);
+ extern bool NextCopyFrom(CopyState cstate, Datum *values, bool *nulls, Oid *oid);
+
extern DestReceiver *CreateCopyDestReceiver(void);
#endif /* COPY_H */