PostgreSQL Source Code  git master
copyto.c File Reference
#include "postgres.h"
#include <ctype.h>
#include <unistd.h>
#include <sys/stat.h>
#include "access/tableam.h"
#include "commands/copy.h"
#include "commands/progress.h"
#include "executor/execdesc.h"
#include "executor/executor.h"
#include "executor/tuptable.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/fd.h"
#include "tcop/tcopprot.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
Include dependency graph for copyto.c:

Go to the source code of this file.

Data Structures

struct  CopyToStateData
 
struct  DR_copy
 

Macros

#define DUMPSOFAR()
 

Typedefs

typedef enum CopyDest CopyDest
 
typedef struct CopyToStateData CopyToStateData
 

Enumerations

enum  CopyDest { COPY_FILE , COPY_FRONTEND , COPY_CALLBACK }
 

Functions

static void EndCopy (CopyToState cstate)
 
static void ClosePipeToProgram (CopyToState cstate)
 
static void CopyOneRowTo (CopyToState cstate, TupleTableSlot *slot)
 
static void CopyAttributeOutText (CopyToState cstate, const char *string)
 
static void CopyAttributeOutCSV (CopyToState cstate, const char *string, bool use_quote)
 
static void SendCopyBegin (CopyToState cstate)
 
static void SendCopyEnd (CopyToState cstate)
 
static void CopySendData (CopyToState cstate, const void *databuf, int datasize)
 
static void CopySendString (CopyToState cstate, const char *str)
 
static void CopySendChar (CopyToState cstate, char c)
 
static void CopySendEndOfRow (CopyToState cstate)
 
static void CopySendInt32 (CopyToState cstate, int32 val)
 
static void CopySendInt16 (CopyToState cstate, int16 val)
 
CopyToState BeginCopyTo (ParseState *pstate, Relation rel, RawStmt *raw_query, Oid queryRelId, const char *filename, bool is_program, copy_data_dest_cb data_dest_cb, List *attnamelist, List *options)
 
void EndCopyTo (CopyToState cstate)
 
uint64 DoCopyTo (CopyToState cstate)
 
static void copy_dest_startup (DestReceiver *self, int operation, TupleDesc typeinfo)
 
static bool copy_dest_receive (TupleTableSlot *slot, DestReceiver *self)
 
static void copy_dest_shutdown (DestReceiver *self)
 
static void copy_dest_destroy (DestReceiver *self)
 
DestReceiverCreateCopyDestReceiver (void)
 

Variables

static const char BinarySignature [11] = "PGCOPY\n\377\r\n\0"
 

Macro Definition Documentation

◆ DUMPSOFAR

#define DUMPSOFAR ( )
Value:
do { \
if (ptr > start) \
CopySendData(cstate, start, ptr - start); \
} while (0)
return str start

Definition at line 975 of file copyto.c.

Typedef Documentation

◆ CopyDest

typedef enum CopyDest CopyDest

◆ CopyToStateData

Enumeration Type Documentation

◆ CopyDest

enum CopyDest
Enumerator
COPY_FILE 
COPY_FRONTEND 
COPY_CALLBACK 

Definition at line 43 of file copyto.c.

44 {
45  COPY_FILE, /* to file (or a piped program) */
46  COPY_FRONTEND, /* to frontend */
47  COPY_CALLBACK, /* to callback function */
48 } CopyDest;
CopyDest
Definition: copyto.c:44
@ COPY_FILE
Definition: copyto.c:45
@ COPY_CALLBACK
Definition: copyto.c:47
@ COPY_FRONTEND
Definition: copyto.c:46

Function Documentation

◆ BeginCopyTo()

CopyToState BeginCopyTo ( ParseState pstate,
Relation  rel,
RawStmt raw_query,
Oid  queryRelId,
const char *  filename,
bool  is_program,
copy_data_dest_cb  data_dest_cb,
List attnamelist,
List options 
)

Definition at line 350 of file copyto.c.

359 {
360  CopyToState cstate;
361  bool pipe = (filename == NULL && data_dest_cb == NULL);
362  TupleDesc tupDesc;
363  int num_phys_attrs;
364  MemoryContext oldcontext;
365  const int progress_cols[] = {
368  };
369  int64 progress_vals[] = {
371  0
372  };
373 
374  if (rel != NULL && rel->rd_rel->relkind != RELKIND_RELATION)
375  {
376  if (rel->rd_rel->relkind == RELKIND_VIEW)
377  ereport(ERROR,
378  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
379  errmsg("cannot copy from view \"%s\"",
381  errhint("Try the COPY (SELECT ...) TO variant.")));
382  else if (rel->rd_rel->relkind == RELKIND_MATVIEW)
383  ereport(ERROR,
384  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
385  errmsg("cannot copy from materialized view \"%s\"",
387  errhint("Try the COPY (SELECT ...) TO variant.")));
388  else if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
389  ereport(ERROR,
390  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
391  errmsg("cannot copy from foreign table \"%s\"",
393  errhint("Try the COPY (SELECT ...) TO variant.")));
394  else if (rel->rd_rel->relkind == RELKIND_SEQUENCE)
395  ereport(ERROR,
396  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
397  errmsg("cannot copy from sequence \"%s\"",
398  RelationGetRelationName(rel))));
399  else if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
400  ereport(ERROR,
401  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
402  errmsg("cannot copy from partitioned table \"%s\"",
404  errhint("Try the COPY (SELECT ...) TO variant.")));
405  else
406  ereport(ERROR,
407  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
408  errmsg("cannot copy from non-table relation \"%s\"",
409  RelationGetRelationName(rel))));
410  }
411 
412 
413  /* Allocate workspace and zero all fields */
414  cstate = (CopyToStateData *) palloc0(sizeof(CopyToStateData));
415 
416  /*
417  * We allocate everything used by a cstate in a new memory context. This
418  * avoids memory leaks during repeated use of COPY in a query.
419  */
421  "COPY",
423 
424  oldcontext = MemoryContextSwitchTo(cstate->copycontext);
425 
426  /* Extract options from the statement node tree */
427  ProcessCopyOptions(pstate, &cstate->opts, false /* is_from */ , options);
428 
429  /* Process the source/target relation or query */
430  if (rel)
431  {
432  Assert(!raw_query);
433 
434  cstate->rel = rel;
435 
436  tupDesc = RelationGetDescr(cstate->rel);
437  }
438  else
439  {
440  List *rewritten;
441  Query *query;
442  PlannedStmt *plan;
444 
445  cstate->rel = NULL;
446 
447  /*
448  * Run parse analysis and rewrite. Note this also acquires sufficient
449  * locks on the source table(s).
450  */
451  rewritten = pg_analyze_and_rewrite_fixedparams(raw_query,
452  pstate->p_sourcetext, NULL, 0,
453  NULL);
454 
455  /* check that we got back something we can work with */
456  if (rewritten == NIL)
457  {
458  ereport(ERROR,
459  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
460  errmsg("DO INSTEAD NOTHING rules are not supported for COPY")));
461  }
462  else if (list_length(rewritten) > 1)
463  {
464  ListCell *lc;
465 
466  /* examine queries to determine which error message to issue */
467  foreach(lc, rewritten)
468  {
469  Query *q = lfirst_node(Query, lc);
470 
471  if (q->querySource == QSRC_QUAL_INSTEAD_RULE)
472  ereport(ERROR,
473  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
474  errmsg("conditional DO INSTEAD rules are not supported for COPY")));
475  if (q->querySource == QSRC_NON_INSTEAD_RULE)
476  ereport(ERROR,
477  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
478  errmsg("DO ALSO rules are not supported for the COPY")));
479  }
480 
481  ereport(ERROR,
482  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
483  errmsg("multi-statement DO INSTEAD rules are not supported for COPY")));
484  }
485 
486  query = linitial_node(Query, rewritten);
487 
488  /* The grammar allows SELECT INTO, but we don't support that */
489  if (query->utilityStmt != NULL &&
491  ereport(ERROR,
492  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
493  errmsg("COPY (SELECT INTO) is not supported")));
494 
495  Assert(query->utilityStmt == NULL);
496 
497  /*
498  * Similarly the grammar doesn't enforce the presence of a RETURNING
499  * clause, but this is required here.
500  */
501  if (query->commandType != CMD_SELECT &&
502  query->returningList == NIL)
503  {
504  Assert(query->commandType == CMD_INSERT ||
505  query->commandType == CMD_UPDATE ||
506  query->commandType == CMD_DELETE ||
507  query->commandType == CMD_MERGE);
508 
509  ereport(ERROR,
510  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
511  errmsg("COPY query must have a RETURNING clause")));
512  }
513 
514  /* plan the query */
515  plan = pg_plan_query(query, pstate->p_sourcetext,
516  CURSOR_OPT_PARALLEL_OK, NULL);
517 
518  /*
519  * With row-level security and a user using "COPY relation TO", we
520  * have to convert the "COPY relation TO" to a query-based COPY (eg:
521  * "COPY (SELECT * FROM ONLY relation) TO"), to allow the rewriter to
522  * add in any RLS clauses.
523  *
524  * When this happens, we are passed in the relid of the originally
525  * found relation (which we have locked). As the planner will look up
526  * the relation again, we double-check here to make sure it found the
527  * same one that we have locked.
528  */
529  if (queryRelId != InvalidOid)
530  {
531  /*
532  * Note that with RLS involved there may be multiple relations,
533  * and while the one we need is almost certainly first, we don't
534  * make any guarantees of that in the planner, so check the whole
535  * list and make sure we find the original relation.
536  */
537  if (!list_member_oid(plan->relationOids, queryRelId))
538  ereport(ERROR,
539  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
540  errmsg("relation referenced by COPY statement has changed")));
541  }
542 
543  /*
544  * Use a snapshot with an updated command ID to ensure this query sees
545  * results of any previously executed queries.
546  */
549 
550  /* Create dest receiver for COPY OUT */
552  ((DR_copy *) dest)->cstate = cstate;
553 
554  /* Create a QueryDesc requesting no output */
555  cstate->queryDesc = CreateQueryDesc(plan, pstate->p_sourcetext,
558  dest, NULL, NULL, 0);
559 
560  /*
561  * Call ExecutorStart to prepare the plan for execution.
562  *
563  * ExecutorStart computes a result tupdesc for us
564  */
565  ExecutorStart(cstate->queryDesc, 0);
566 
567  tupDesc = cstate->queryDesc->tupDesc;
568  }
569 
570  /* Generate or convert list of attributes to process */
571  cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
572 
573  num_phys_attrs = tupDesc->natts;
574 
575  /* Convert FORCE_QUOTE name list to per-column flags, check validity */
576  cstate->opts.force_quote_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
577  if (cstate->opts.force_quote_all)
578  {
579  MemSet(cstate->opts.force_quote_flags, true, num_phys_attrs * sizeof(bool));
580  }
581  else if (cstate->opts.force_quote)
582  {
583  List *attnums;
584  ListCell *cur;
585 
586  attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_quote);
587 
588  foreach(cur, attnums)
589  {
590  int attnum = lfirst_int(cur);
591  Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
592 
593  if (!list_member_int(cstate->attnumlist, attnum))
594  ereport(ERROR,
595  (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
596  errmsg("FORCE_QUOTE column \"%s\" not referenced by COPY",
597  NameStr(attr->attname))));
598  cstate->opts.force_quote_flags[attnum - 1] = true;
599  }
600  }
601 
602  /* Use client encoding when ENCODING option is not specified. */
603  if (cstate->opts.file_encoding < 0)
605  else
606  cstate->file_encoding = cstate->opts.file_encoding;
607 
608  /*
609  * Set up encoding conversion info if the file and server encodings differ
610  * (see also pg_server_to_any).
611  */
612  if (cstate->file_encoding == GetDatabaseEncoding() ||
613  cstate->file_encoding == PG_SQL_ASCII)
614  cstate->need_transcoding = false;
615  else
616  cstate->need_transcoding = true;
617 
618  /* See Multibyte encoding comment above */
620 
621  cstate->copy_dest = COPY_FILE; /* default */
622 
623  if (data_dest_cb)
624  {
625  progress_vals[1] = PROGRESS_COPY_TYPE_CALLBACK;
626  cstate->copy_dest = COPY_CALLBACK;
627  cstate->data_dest_cb = data_dest_cb;
628  }
629  else if (pipe)
630  {
631  progress_vals[1] = PROGRESS_COPY_TYPE_PIPE;
632 
633  Assert(!is_program); /* the grammar does not allow this */
635  cstate->copy_file = stdout;
636  }
637  else
638  {
639  cstate->filename = pstrdup(filename);
640  cstate->is_program = is_program;
641 
642  if (is_program)
643  {
644  progress_vals[1] = PROGRESS_COPY_TYPE_PROGRAM;
645  cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_W);
646  if (cstate->copy_file == NULL)
647  ereport(ERROR,
649  errmsg("could not execute command \"%s\": %m",
650  cstate->filename)));
651  }
652  else
653  {
654  mode_t oumask; /* Pre-existing umask value */
655  struct stat st;
656 
657  progress_vals[1] = PROGRESS_COPY_TYPE_FILE;
658 
659  /*
660  * Prevent write to relative path ... too easy to shoot oneself in
661  * the foot by overwriting a database file ...
662  */
664  ereport(ERROR,
665  (errcode(ERRCODE_INVALID_NAME),
666  errmsg("relative path not allowed for COPY to file")));
667 
668  oumask = umask(S_IWGRP | S_IWOTH);
669  PG_TRY();
670  {
671  cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_W);
672  }
673  PG_FINALLY();
674  {
675  umask(oumask);
676  }
677  PG_END_TRY();
678  if (cstate->copy_file == NULL)
679  {
680  /* copy errno because ereport subfunctions might change it */
681  int save_errno = errno;
682 
683  ereport(ERROR,
685  errmsg("could not open file \"%s\" for writing: %m",
686  cstate->filename),
687  (save_errno == ENOENT || save_errno == EACCES) ?
688  errhint("COPY TO instructs the PostgreSQL server process to write a file. "
689  "You may want a client-side facility such as psql's \\copy.") : 0));
690  }
691 
692  if (fstat(fileno(cstate->copy_file), &st))
693  ereport(ERROR,
695  errmsg("could not stat file \"%s\": %m",
696  cstate->filename)));
697 
698  if (S_ISDIR(st.st_mode))
699  ereport(ERROR,
700  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
701  errmsg("\"%s\" is a directory", cstate->filename)));
702  }
703  }
704 
705  /* initialize progress */
707  cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
708  pgstat_progress_update_multi_param(2, progress_cols, progress_vals);
709 
710  cstate->bytes_processed = 0;
711 
712  MemoryContextSwitchTo(oldcontext);
713 
714  return cstate;
715 }
List * CopyGetAttnums(TupleDesc tupDesc, Relation rel, List *attnamelist)
Definition: copy.c:865
void ProcessCopyOptions(ParseState *pstate, CopyFormatOptions *opts_out, bool is_from, List *options)
Definition: copy.c:466
void pgstat_progress_start_command(ProgressCommandType cmdtype, Oid relid)
void pgstat_progress_update_multi_param(int nparam, const int *index, const int64 *val)
@ PROGRESS_COMMAND_COPY
#define NameStr(name)
Definition: c.h:746
#define Assert(condition)
Definition: c.h:858
#define PG_BINARY_W
Definition: c.h:1276
#define MemSet(start, val, len)
Definition: c.h:1020
DestReceiver * CreateDestReceiver(CommandDest dest)
Definition: dest.c:113
@ DestRemote
Definition: dest.h:89
@ DestCopyOut
Definition: dest.h:95
struct cursor * cur
Definition: ecpg.c:28
int errcode_for_file_access(void)
Definition: elog.c:882
int errhint(const char *fmt,...)
Definition: elog.c:1319
int errcode(int sqlerrcode)
Definition: elog.c:859
int errmsg(const char *fmt,...)
Definition: elog.c:1072
#define PG_TRY(...)
Definition: elog.h:370
#define PG_END_TRY(...)
Definition: elog.h:395
#define ERROR
Definition: elog.h:39
#define PG_FINALLY(...)
Definition: elog.h:387
#define ereport(elevel,...)
Definition: elog.h:149
void ExecutorStart(QueryDesc *queryDesc, int eflags)
Definition: execMain.c:124
FILE * AllocateFile(const char *name, const char *mode)
Definition: fd.c:2583
FILE * OpenPipeStream(const char *command, const char *mode)
Definition: fd.c:2686
bool list_member_int(const List *list, int datum)
Definition: list.c:702
bool list_member_oid(const List *list, Oid datum)
Definition: list.c:722
int GetDatabaseEncoding(void)
Definition: mbutils.c:1261
int pg_get_client_encoding(void)
Definition: mbutils.c:336
char * pstrdup(const char *in)
Definition: mcxt.c:1695
void * palloc0(Size size)
Definition: mcxt.c:1346
MemoryContext CurrentMemoryContext
Definition: mcxt.c:143
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:160
#define IsA(nodeptr, _type_)
Definition: nodes.h:158
@ CMD_MERGE
Definition: nodes.h:269
@ CMD_INSERT
Definition: nodes.h:267
@ CMD_DELETE
Definition: nodes.h:268
@ CMD_UPDATE
Definition: nodes.h:266
@ CMD_SELECT
Definition: nodes.h:265
@ QSRC_NON_INSTEAD_RULE
Definition: parsenodes.h:40
@ QSRC_QUAL_INSTEAD_RULE
Definition: parsenodes.h:39
#define CURSOR_OPT_PARALLEL_OK
Definition: parsenodes.h:3300
int16 attnum
Definition: pg_attribute.h:74
FormData_pg_attribute * Form_pg_attribute
Definition: pg_attribute.h:209
static char * filename
Definition: pg_dumpall.c:119
#define lfirst_node(type, lc)
Definition: pg_list.h:176
static int list_length(const List *l)
Definition: pg_list.h:152
#define linitial_node(type, l)
Definition: pg_list.h:181
#define NIL
Definition: pg_list.h:68
#define lfirst_int(lc)
Definition: pg_list.h:173
#define plan(x)
Definition: pg_regress.c:162
@ PG_SQL_ASCII
Definition: pg_wchar.h:226
#define PG_ENCODING_IS_CLIENT_ONLY(_enc)
Definition: pg_wchar.h:284
#define is_absolute_path(filename)
Definition: port.h:103
CommandDest whereToSendOutput
Definition: postgres.c:90
List * pg_analyze_and_rewrite_fixedparams(RawStmt *parsetree, const char *query_string, const Oid *paramTypes, int numParams, QueryEnvironment *queryEnv)
Definition: postgres.c:671
PlannedStmt * pg_plan_query(Query *querytree, const char *query_string, int cursorOptions, ParamListInfo boundParams)
Definition: postgres.c:886
#define InvalidOid
Definition: postgres_ext.h:36
QueryDesc * CreateQueryDesc(PlannedStmt *plannedstmt, const char *sourceText, Snapshot snapshot, Snapshot crosscheck_snapshot, DestReceiver *dest, ParamListInfo params, QueryEnvironment *queryEnv, int instrument_options)
Definition: pquery.c:67
#define PROGRESS_COPY_COMMAND
Definition: progress.h:143
#define PROGRESS_COPY_TYPE_FILE
Definition: progress.h:152
#define PROGRESS_COPY_COMMAND_TO
Definition: progress.h:149
#define PROGRESS_COPY_TYPE
Definition: progress.h:144
#define PROGRESS_COPY_TYPE_PROGRAM
Definition: progress.h:153
#define PROGRESS_COPY_TYPE_CALLBACK
Definition: progress.h:155
#define PROGRESS_COPY_TYPE_PIPE
Definition: progress.h:154
MemoryContextSwitchTo(old_ctx)
#define RelationGetRelid(relation)
Definition: rel.h:505
#define RelationGetDescr(relation)
Definition: rel.h:531
#define RelationGetRelationName(relation)
Definition: rel.h:539
void UpdateActiveSnapshotCommandId(void)
Definition: snapmgr.c:712
void PushCopiedSnapshot(Snapshot snapshot)
Definition: snapmgr.c:700
Snapshot GetActiveSnapshot(void)
Definition: snapmgr.c:770
#define InvalidSnapshot
Definition: snapshot.h:123
bool force_quote_all
Definition: copy.h:75
List * force_quote
Definition: copy.h:74
bool * force_quote_flags
Definition: copy.h:76
int file_encoding
Definition: copy.h:60
MemoryContext copycontext
Definition: copyto.c:90
Relation rel
Definition: copyto.c:77
copy_data_dest_cb data_dest_cb
Definition: copyto.c:82
bool encoding_embeds_ascii
Definition: copyto.c:74
CopyDest copy_dest
Definition: copyto.c:68
bool need_transcoding
Definition: copyto.c:73
bool is_program
Definition: copyto.c:81
FILE * copy_file
Definition: copyto.c:69
int file_encoding
Definition: copyto.c:72
CopyFormatOptions opts
Definition: copyto.c:84
uint64 bytes_processed
Definition: copyto.c:94
char * filename
Definition: copyto.c:80
List * attnumlist
Definition: copyto.c:79
QueryDesc * queryDesc
Definition: copyto.c:78
Definition: copyto.c:99
Definition: pg_list.h:54
const char * p_sourcetext
Definition: parse_node.h:193
TupleDesc tupDesc
Definition: execdesc.h:47
List * returningList
Definition: parsenodes.h:198
CmdType commandType
Definition: parsenodes.h:121
Node * utilityStmt
Definition: parsenodes.h:136
Form_pg_class rd_rel
Definition: rel.h:111
#define TupleDescAttr(tupdesc, i)
Definition: tupdesc.h:92
#define S_IWOTH
Definition: win32_port.h:316
#define S_ISDIR(m)
Definition: win32_port.h:325
#define fstat
Definition: win32_port.h:283
#define S_IWGRP
Definition: win32_port.h:304

References AllocateFile(), ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, attnum, CopyToStateData::attnumlist, CopyToStateData::bytes_processed, CMD_DELETE, CMD_INSERT, CMD_MERGE, CMD_SELECT, CMD_UPDATE, Query::commandType, COPY_CALLBACK, CopyToStateData::copy_dest, COPY_FILE, CopyToStateData::copy_file, CopyToStateData::copycontext, CopyGetAttnums(), CreateDestReceiver(), CreateQueryDesc(), cur, CurrentMemoryContext, CURSOR_OPT_PARALLEL_OK, CopyToStateData::data_dest_cb, generate_unaccent_rules::dest, DestCopyOut, DestRemote, CopyToStateData::encoding_embeds_ascii, ereport, errcode(), errcode_for_file_access(), errhint(), errmsg(), ERROR, ExecutorStart(), CopyToStateData::file_encoding, CopyFormatOptions::file_encoding, CopyToStateData::filename, filename, CopyFormatOptions::force_quote, CopyFormatOptions::force_quote_all, CopyFormatOptions::force_quote_flags, fstat, GetActiveSnapshot(), GetDatabaseEncoding(), InvalidOid, InvalidSnapshot, is_absolute_path, CopyToStateData::is_program, IsA, lfirst_int, lfirst_node, linitial_node, list_length(), list_member_int(), list_member_oid(), MemoryContextSwitchTo(), MemSet, NameStr, TupleDescData::natts, CopyToStateData::need_transcoding, NIL, OpenPipeStream(), CopyToStateData::opts, ParseState::p_sourcetext, palloc0(), pg_analyze_and_rewrite_fixedparams(), PG_BINARY_W, PG_ENCODING_IS_CLIENT_ONLY, PG_END_TRY, PG_FINALLY, pg_get_client_encoding(), pg_plan_query(), PG_SQL_ASCII, PG_TRY, pgstat_progress_start_command(), pgstat_progress_update_multi_param(), plan, ProcessCopyOptions(), PROGRESS_COMMAND_COPY, PROGRESS_COPY_COMMAND, PROGRESS_COPY_COMMAND_TO, PROGRESS_COPY_TYPE, PROGRESS_COPY_TYPE_CALLBACK, PROGRESS_COPY_TYPE_FILE, PROGRESS_COPY_TYPE_PIPE, PROGRESS_COPY_TYPE_PROGRAM, pstrdup(), PushCopiedSnapshot(), QSRC_NON_INSTEAD_RULE, QSRC_QUAL_INSTEAD_RULE, CopyToStateData::queryDesc, RelationData::rd_rel, CopyToStateData::rel, RelationGetDescr, RelationGetRelationName, RelationGetRelid, Query::returningList, S_ISDIR, S_IWGRP, S_IWOTH, stat::st_mode, generate_unaccent_rules::stdout, QueryDesc::tupDesc, TupleDescAttr, UpdateActiveSnapshotCommandId(), Query::utilityStmt, and whereToSendOutput.

Referenced by DoCopy(), and test_copy_to_callback().

◆ ClosePipeToProgram()

static void ClosePipeToProgram ( CopyToState  cstate)
static

Definition at line 289 of file copyto.c.

290 {
291  int pclose_rc;
292 
293  Assert(cstate->is_program);
294 
295  pclose_rc = ClosePipeStream(cstate->copy_file);
296  if (pclose_rc == -1)
297  ereport(ERROR,
299  errmsg("could not close pipe to external command: %m")));
300  else if (pclose_rc != 0)
301  {
302  ereport(ERROR,
303  (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
304  errmsg("program \"%s\" failed",
305  cstate->filename),
306  errdetail_internal("%s", wait_result_to_str(pclose_rc))));
307  }
308 }
int errdetail_internal(const char *fmt,...)
Definition: elog.c:1232
int ClosePipeStream(FILE *file)
Definition: fd.c:2991
char * wait_result_to_str(int exitstatus)
Definition: wait_error.c:33

References Assert, ClosePipeStream(), CopyToStateData::copy_file, ereport, errcode(), errcode_for_file_access(), errdetail_internal(), errmsg(), ERROR, CopyToStateData::filename, CopyToStateData::is_program, and wait_result_to_str().

Referenced by CopySendEndOfRow(), and EndCopy().

◆ copy_dest_destroy()

static void copy_dest_destroy ( DestReceiver self)
static

Definition at line 1258 of file copyto.c.

1259 {
1260  pfree(self);
1261 }
void pfree(void *pointer)
Definition: mcxt.c:1520

References pfree().

Referenced by CreateCopyDestReceiver().

◆ copy_dest_receive()

static bool copy_dest_receive ( TupleTableSlot slot,
DestReceiver self 
)
static

Definition at line 1230 of file copyto.c.

1231 {
1232  DR_copy *myState = (DR_copy *) self;
1233  CopyToState cstate = myState->cstate;
1234 
1235  /* Send the data */
1236  CopyOneRowTo(cstate, slot);
1237 
1238  /* Increment the number of processed tuples, and report the progress */
1240  ++myState->processed);
1241 
1242  return true;
1243 }
void pgstat_progress_update_param(int index, int64 val)
static void CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
Definition: copyto.c:902
#define PROGRESS_COPY_TUPLES_PROCESSED
Definition: progress.h:141
CopyToState cstate
Definition: copyto.c:101
uint64 processed
Definition: copyto.c:102

References CopyOneRowTo(), DR_copy::cstate, pgstat_progress_update_param(), DR_copy::processed, and PROGRESS_COPY_TUPLES_PROCESSED.

Referenced by CreateCopyDestReceiver().

◆ copy_dest_shutdown()

static void copy_dest_shutdown ( DestReceiver self)
static

Definition at line 1249 of file copyto.c.

1250 {
1251  /* no-op */
1252 }

Referenced by CreateCopyDestReceiver().

◆ copy_dest_startup()

static void copy_dest_startup ( DestReceiver self,
int  operation,
TupleDesc  typeinfo 
)
static

Definition at line 1221 of file copyto.c.

1222 {
1223  /* no-op */
1224 }

Referenced by CreateCopyDestReceiver().

◆ CopyAttributeOutCSV()

static void CopyAttributeOutCSV ( CopyToState  cstate,
const char *  string,
bool  use_quote 
)
static

Definition at line 1135 of file copyto.c.

1137 {
1138  const char *ptr;
1139  const char *start;
1140  char c;
1141  char delimc = cstate->opts.delim[0];
1142  char quotec = cstate->opts.quote[0];
1143  char escapec = cstate->opts.escape[0];
1144  bool single_attr = (list_length(cstate->attnumlist) == 1);
1145 
1146  /* force quoting if it matches null_print (before conversion!) */
1147  if (!use_quote && strcmp(string, cstate->opts.null_print) == 0)
1148  use_quote = true;
1149 
1150  if (cstate->need_transcoding)
1151  ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding);
1152  else
1153  ptr = string;
1154 
1155  /*
1156  * Make a preliminary pass to discover if it needs quoting
1157  */
1158  if (!use_quote)
1159  {
1160  /*
1161  * Because '\.' can be a data value, quote it if it appears alone on a
1162  * line so it is not interpreted as the end-of-data marker.
1163  */
1164  if (single_attr && strcmp(ptr, "\\.") == 0)
1165  use_quote = true;
1166  else
1167  {
1168  const char *tptr = ptr;
1169 
1170  while ((c = *tptr) != '\0')
1171  {
1172  if (c == delimc || c == quotec || c == '\n' || c == '\r')
1173  {
1174  use_quote = true;
1175  break;
1176  }
1177  if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii)
1178  tptr += pg_encoding_mblen(cstate->file_encoding, tptr);
1179  else
1180  tptr++;
1181  }
1182  }
1183  }
1184 
1185  if (use_quote)
1186  {
1187  CopySendChar(cstate, quotec);
1188 
1189  /*
1190  * We adopt the same optimization strategy as in CopyAttributeOutText
1191  */
1192  start = ptr;
1193  while ((c = *ptr) != '\0')
1194  {
1195  if (c == quotec || c == escapec)
1196  {
1197  DUMPSOFAR();
1198  CopySendChar(cstate, escapec);
1199  start = ptr; /* we include char in next run */
1200  }
1201  if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii)
1202  ptr += pg_encoding_mblen(cstate->file_encoding, ptr);
1203  else
1204  ptr++;
1205  }
1206  DUMPSOFAR();
1207 
1208  CopySendChar(cstate, quotec);
1209  }
1210  else
1211  {
1212  /* If it doesn't need quoting, we can just dump it as-is */
1213  CopySendString(cstate, ptr);
1214  }
1215 }
#define IS_HIGHBIT_SET(ch)
Definition: c.h:1155
#define DUMPSOFAR()
Definition: copyto.c:975
static void CopySendChar(CopyToState cstate, char c)
Definition: copyto.c:181
static void CopySendString(CopyToState cstate, const char *str)
Definition: copyto.c:175
char * pg_server_to_any(const char *s, int len, int encoding)
Definition: mbutils.c:749
char * c
char string[11]
Definition: preproc-type.c:52
char * quote
Definition: copy.h:72
char * escape
Definition: copy.h:73
char * null_print
Definition: copy.h:66
char * delim
Definition: copy.h:71
int pg_encoding_mblen(int encoding, const char *mbstr)
Definition: wchar.c:2069

References CopyToStateData::attnumlist, CopySendChar(), CopySendString(), CopyFormatOptions::delim, DUMPSOFAR, CopyToStateData::encoding_embeds_ascii, CopyFormatOptions::escape, CopyToStateData::file_encoding, IS_HIGHBIT_SET, list_length(), CopyToStateData::need_transcoding, CopyFormatOptions::null_print, CopyToStateData::opts, pg_encoding_mblen(), pg_server_to_any(), CopyFormatOptions::quote, and start.

Referenced by CopyOneRowTo(), and DoCopyTo().

◆ CopyAttributeOutText()

static void CopyAttributeOutText ( CopyToState  cstate,
const char *  string 
)
static

Definition at line 982 of file copyto.c.

983 {
984  const char *ptr;
985  const char *start;
986  char c;
987  char delimc = cstate->opts.delim[0];
988 
989  if (cstate->need_transcoding)
990  ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding);
991  else
992  ptr = string;
993 
994  /*
995  * We have to grovel through the string searching for control characters
996  * and instances of the delimiter character. In most cases, though, these
997  * are infrequent. To avoid overhead from calling CopySendData once per
998  * character, we dump out all characters between escaped characters in a
999  * single call. The loop invariant is that the data from "start" to "ptr"
1000  * can be sent literally, but hasn't yet been.
1001  *
1002  * We can skip pg_encoding_mblen() overhead when encoding is safe, because
1003  * in valid backend encodings, extra bytes of a multibyte character never
1004  * look like ASCII. This loop is sufficiently performance-critical that
1005  * it's worth making two copies of it to get the IS_HIGHBIT_SET() test out
1006  * of the normal safe-encoding path.
1007  */
1008  if (cstate->encoding_embeds_ascii)
1009  {
1010  start = ptr;
1011  while ((c = *ptr) != '\0')
1012  {
1013  if ((unsigned char) c < (unsigned char) 0x20)
1014  {
1015  /*
1016  * \r and \n must be escaped, the others are traditional. We
1017  * prefer to dump these using the C-like notation, rather than
1018  * a backslash and the literal character, because it makes the
1019  * dump file a bit more proof against Microsoftish data
1020  * mangling.
1021  */
1022  switch (c)
1023  {
1024  case '\b':
1025  c = 'b';
1026  break;
1027  case '\f':
1028  c = 'f';
1029  break;
1030  case '\n':
1031  c = 'n';
1032  break;
1033  case '\r':
1034  c = 'r';
1035  break;
1036  case '\t':
1037  c = 't';
1038  break;
1039  case '\v':
1040  c = 'v';
1041  break;
1042  default:
1043  /* If it's the delimiter, must backslash it */
1044  if (c == delimc)
1045  break;
1046  /* All ASCII control chars are length 1 */
1047  ptr++;
1048  continue; /* fall to end of loop */
1049  }
1050  /* if we get here, we need to convert the control char */
1051  DUMPSOFAR();
1052  CopySendChar(cstate, '\\');
1053  CopySendChar(cstate, c);
1054  start = ++ptr; /* do not include char in next run */
1055  }
1056  else if (c == '\\' || c == delimc)
1057  {
1058  DUMPSOFAR();
1059  CopySendChar(cstate, '\\');
1060  start = ptr++; /* we include char in next run */
1061  }
1062  else if (IS_HIGHBIT_SET(c))
1063  ptr += pg_encoding_mblen(cstate->file_encoding, ptr);
1064  else
1065  ptr++;
1066  }
1067  }
1068  else
1069  {
1070  start = ptr;
1071  while ((c = *ptr) != '\0')
1072  {
1073  if ((unsigned char) c < (unsigned char) 0x20)
1074  {
1075  /*
1076  * \r and \n must be escaped, the others are traditional. We
1077  * prefer to dump these using the C-like notation, rather than
1078  * a backslash and the literal character, because it makes the
1079  * dump file a bit more proof against Microsoftish data
1080  * mangling.
1081  */
1082  switch (c)
1083  {
1084  case '\b':
1085  c = 'b';
1086  break;
1087  case '\f':
1088  c = 'f';
1089  break;
1090  case '\n':
1091  c = 'n';
1092  break;
1093  case '\r':
1094  c = 'r';
1095  break;
1096  case '\t':
1097  c = 't';
1098  break;
1099  case '\v':
1100  c = 'v';
1101  break;
1102  default:
1103  /* If it's the delimiter, must backslash it */
1104  if (c == delimc)
1105  break;
1106  /* All ASCII control chars are length 1 */
1107  ptr++;
1108  continue; /* fall to end of loop */
1109  }
1110  /* if we get here, we need to convert the control char */
1111  DUMPSOFAR();
1112  CopySendChar(cstate, '\\');
1113  CopySendChar(cstate, c);
1114  start = ++ptr; /* do not include char in next run */
1115  }
1116  else if (c == '\\' || c == delimc)
1117  {
1118  DUMPSOFAR();
1119  CopySendChar(cstate, '\\');
1120  start = ptr++; /* we include char in next run */
1121  }
1122  else
1123  ptr++;
1124  }
1125  }
1126 
1127  DUMPSOFAR();
1128 }

References CopySendChar(), CopyFormatOptions::delim, DUMPSOFAR, CopyToStateData::encoding_embeds_ascii, CopyToStateData::file_encoding, IS_HIGHBIT_SET, CopyToStateData::need_transcoding, CopyToStateData::opts, pg_encoding_mblen(), pg_server_to_any(), and start.

Referenced by CopyOneRowTo(), and DoCopyTo().

◆ CopyOneRowTo()

static void CopyOneRowTo ( CopyToState  cstate,
TupleTableSlot slot 
)
static

Definition at line 902 of file copyto.c.

903 {
904  bool need_delim = false;
905  FmgrInfo *out_functions = cstate->out_functions;
906  MemoryContext oldcontext;
907  ListCell *cur;
908  char *string;
909 
911  oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
912 
913  if (cstate->opts.binary)
914  {
915  /* Binary per-tuple header */
916  CopySendInt16(cstate, list_length(cstate->attnumlist));
917  }
918 
919  /* Make sure the tuple is fully deconstructed */
920  slot_getallattrs(slot);
921 
922  foreach(cur, cstate->attnumlist)
923  {
924  int attnum = lfirst_int(cur);
925  Datum value = slot->tts_values[attnum - 1];
926  bool isnull = slot->tts_isnull[attnum - 1];
927 
928  if (!cstate->opts.binary)
929  {
930  if (need_delim)
931  CopySendChar(cstate, cstate->opts.delim[0]);
932  need_delim = true;
933  }
934 
935  if (isnull)
936  {
937  if (!cstate->opts.binary)
938  CopySendString(cstate, cstate->opts.null_print_client);
939  else
940  CopySendInt32(cstate, -1);
941  }
942  else
943  {
944  if (!cstate->opts.binary)
945  {
946  string = OutputFunctionCall(&out_functions[attnum - 1],
947  value);
948  if (cstate->opts.csv_mode)
949  CopyAttributeOutCSV(cstate, string,
950  cstate->opts.force_quote_flags[attnum - 1]);
951  else
952  CopyAttributeOutText(cstate, string);
953  }
954  else
955  {
956  bytea *outputbytes;
957 
958  outputbytes = SendFunctionCall(&out_functions[attnum - 1],
959  value);
960  CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
961  CopySendData(cstate, VARDATA(outputbytes),
962  VARSIZE(outputbytes) - VARHDRSZ);
963  }
964  }
965  }
966 
967  CopySendEndOfRow(cstate);
968 
969  MemoryContextSwitchTo(oldcontext);
970 }
#define VARHDRSZ
Definition: c.h:692
static void CopySendInt32(CopyToState cstate, int32 val)
Definition: copyto.c:265
static void CopyAttributeOutCSV(CopyToState cstate, const char *string, bool use_quote)
Definition: copyto.c:1135
static void CopyAttributeOutText(CopyToState cstate, const char *string)
Definition: copyto.c:982
static void CopySendInt16(CopyToState cstate, int16 val)
Definition: copyto.c:277
static void CopySendData(CopyToState cstate, const void *databuf, int datasize)
Definition: copyto.c:169
static void CopySendEndOfRow(CopyToState cstate)
Definition: copyto.c:187
bytea * SendFunctionCall(FmgrInfo *flinfo, Datum val)
Definition: fmgr.c:1744
char * OutputFunctionCall(FmgrInfo *flinfo, Datum val)
Definition: fmgr.c:1683
static struct @155 value
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:383
uintptr_t Datum
Definition: postgres.h:64
bool binary
Definition: copy.h:62
char * null_print_client
Definition: copy.h:68
bool csv_mode
Definition: copy.h:64
FmgrInfo * out_functions
Definition: copyto.c:92
MemoryContext rowcontext
Definition: copyto.c:93
Definition: fmgr.h:57
bool * tts_isnull
Definition: tuptable.h:127
Datum * tts_values
Definition: tuptable.h:125
Definition: c.h:687
static void slot_getallattrs(TupleTableSlot *slot)
Definition: tuptable.h:368
#define VARDATA(PTR)
Definition: varatt.h:278
#define VARSIZE(PTR)
Definition: varatt.h:279

References attnum, CopyToStateData::attnumlist, CopyFormatOptions::binary, CopyAttributeOutCSV(), CopyAttributeOutText(), CopySendChar(), CopySendData(), CopySendEndOfRow(), CopySendInt16(), CopySendInt32(), CopySendString(), CopyFormatOptions::csv_mode, cur, CopyFormatOptions::delim, CopyFormatOptions::force_quote_flags, lfirst_int, list_length(), MemoryContextReset(), MemoryContextSwitchTo(), CopyFormatOptions::null_print_client, CopyToStateData::opts, CopyToStateData::out_functions, OutputFunctionCall(), CopyToStateData::rowcontext, SendFunctionCall(), slot_getallattrs(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_values, value, VARDATA, VARHDRSZ, and VARSIZE.

Referenced by copy_dest_receive(), and DoCopyTo().

◆ CopySendChar()

static void CopySendChar ( CopyToState  cstate,
char  c 
)
static

Definition at line 181 of file copyto.c.

182 {
184 }
#define appendStringInfoCharMacro(str, ch)
Definition: stringinfo.h:204
StringInfo fe_msgbuf
Definition: copyto.c:70

References appendStringInfoCharMacro, and CopyToStateData::fe_msgbuf.

Referenced by CopyAttributeOutCSV(), CopyAttributeOutText(), CopyOneRowTo(), CopySendEndOfRow(), and DoCopyTo().

◆ CopySendData()

static void CopySendData ( CopyToState  cstate,
const void *  databuf,
int  datasize 
)
static

Definition at line 169 of file copyto.c.

170 {
171  appendBinaryStringInfo(cstate->fe_msgbuf, databuf, datasize);
172 }
void appendBinaryStringInfo(StringInfo str, const void *data, int datalen)
Definition: stringinfo.c:233

References appendBinaryStringInfo(), and CopyToStateData::fe_msgbuf.

Referenced by CopyOneRowTo(), CopySendInt16(), CopySendInt32(), and DoCopyTo().

◆ CopySendEndOfRow()

static void CopySendEndOfRow ( CopyToState  cstate)
static

Definition at line 187 of file copyto.c.

188 {
189  StringInfo fe_msgbuf = cstate->fe_msgbuf;
190 
191  switch (cstate->copy_dest)
192  {
193  case COPY_FILE:
194  if (!cstate->opts.binary)
195  {
196  /* Default line termination depends on platform */
197 #ifndef WIN32
198  CopySendChar(cstate, '\n');
199 #else
200  CopySendString(cstate, "\r\n");
201 #endif
202  }
203 
204  if (fwrite(fe_msgbuf->data, fe_msgbuf->len, 1,
205  cstate->copy_file) != 1 ||
206  ferror(cstate->copy_file))
207  {
208  if (cstate->is_program)
209  {
210  if (errno == EPIPE)
211  {
212  /*
213  * The pipe will be closed automatically on error at
214  * the end of transaction, but we might get a better
215  * error message from the subprocess' exit code than
216  * just "Broken Pipe"
217  */
218  ClosePipeToProgram(cstate);
219 
220  /*
221  * If ClosePipeToProgram() didn't throw an error, the
222  * program terminated normally, but closed the pipe
223  * first. Restore errno, and throw an error.
224  */
225  errno = EPIPE;
226  }
227  ereport(ERROR,
229  errmsg("could not write to COPY program: %m")));
230  }
231  else
232  ereport(ERROR,
234  errmsg("could not write to COPY file: %m")));
235  }
236  break;
237  case COPY_FRONTEND:
238  /* The FE/BE protocol uses \n as newline for all platforms */
239  if (!cstate->opts.binary)
240  CopySendChar(cstate, '\n');
241 
242  /* Dump the accumulated row as one CopyData message */
243  (void) pq_putmessage(PqMsg_CopyData, fe_msgbuf->data, fe_msgbuf->len);
244  break;
245  case COPY_CALLBACK:
246  cstate->data_dest_cb(fe_msgbuf->data, fe_msgbuf->len);
247  break;
248  }
249 
250  /* Update the progress */
251  cstate->bytes_processed += fe_msgbuf->len;
253 
254  resetStringInfo(fe_msgbuf);
255 }
static void ClosePipeToProgram(CopyToState cstate)
Definition: copyto.c:289
#define pq_putmessage(msgtype, s, len)
Definition: libpq.h:49
#define PROGRESS_COPY_BYTES_PROCESSED
Definition: progress.h:139
#define PqMsg_CopyData
Definition: protocol.h:65
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:78

References CopyFormatOptions::binary, CopyToStateData::bytes_processed, ClosePipeToProgram(), COPY_CALLBACK, CopyToStateData::copy_dest, COPY_FILE, CopyToStateData::copy_file, COPY_FRONTEND, CopySendChar(), CopySendString(), StringInfoData::data, CopyToStateData::data_dest_cb, ereport, errcode_for_file_access(), errmsg(), ERROR, CopyToStateData::fe_msgbuf, CopyToStateData::is_program, StringInfoData::len, CopyToStateData::opts, pgstat_progress_update_param(), pq_putmessage, PqMsg_CopyData, PROGRESS_COPY_BYTES_PROCESSED, and resetStringInfo().

Referenced by CopyOneRowTo(), and DoCopyTo().

◆ CopySendInt16()

static void CopySendInt16 ( CopyToState  cstate,
int16  val 
)
inlinestatic

Definition at line 277 of file copyto.c.

278 {
279  uint16 buf;
280 
281  buf = pg_hton16((uint16) val);
282  CopySendData(cstate, &buf, sizeof(buf));
283 }
unsigned short uint16
Definition: c.h:505
long val
Definition: informix.c:670
#define pg_hton16(x)
Definition: pg_bswap.h:120
static char * buf
Definition: pg_test_fsync.c:73

References buf, CopySendData(), pg_hton16, and val.

Referenced by CopyOneRowTo(), and DoCopyTo().

◆ CopySendInt32()

static void CopySendInt32 ( CopyToState  cstate,
int32  val 
)
inlinestatic

Definition at line 265 of file copyto.c.

266 {
267  uint32 buf;
268 
269  buf = pg_hton32((uint32) val);
270  CopySendData(cstate, &buf, sizeof(buf));
271 }
unsigned int uint32
Definition: c.h:506
#define pg_hton32(x)
Definition: pg_bswap.h:121

References buf, CopySendData(), pg_hton32, and val.

Referenced by CopyOneRowTo(), and DoCopyTo().

◆ CopySendString()

static void CopySendString ( CopyToState  cstate,
const char *  str 
)
static

Definition at line 175 of file copyto.c.

176 {
177  appendBinaryStringInfo(cstate->fe_msgbuf, str, strlen(str));
178 }
const char * str

References appendBinaryStringInfo(), CopyToStateData::fe_msgbuf, and str.

Referenced by CopyAttributeOutCSV(), CopyOneRowTo(), and CopySendEndOfRow().

◆ CreateCopyDestReceiver()

DestReceiver* CreateCopyDestReceiver ( void  )

Definition at line 1267 of file copyto.c.

1268 {
1269  DR_copy *self = (DR_copy *) palloc(sizeof(DR_copy));
1270 
1271  self->pub.receiveSlot = copy_dest_receive;
1272  self->pub.rStartup = copy_dest_startup;
1273  self->pub.rShutdown = copy_dest_shutdown;
1274  self->pub.rDestroy = copy_dest_destroy;
1275  self->pub.mydest = DestCopyOut;
1276 
1277  self->cstate = NULL; /* will be set later */
1278  self->processed = 0;
1279 
1280  return (DestReceiver *) self;
1281 }
static bool copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
Definition: copyto.c:1230
static void copy_dest_destroy(DestReceiver *self)
Definition: copyto.c:1258
static void copy_dest_shutdown(DestReceiver *self)
Definition: copyto.c:1249
static void copy_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
Definition: copyto.c:1221
void * palloc(Size size)
Definition: mcxt.c:1316

References copy_dest_destroy(), copy_dest_receive(), copy_dest_shutdown(), copy_dest_startup(), DestCopyOut, and palloc().

Referenced by CreateDestReceiver().

◆ DoCopyTo()

uint64 DoCopyTo ( CopyToState  cstate)

Definition at line 742 of file copyto.c.

743 {
744  bool pipe = (cstate->filename == NULL && cstate->data_dest_cb == NULL);
745  bool fe_copy = (pipe && whereToSendOutput == DestRemote);
746  TupleDesc tupDesc;
747  int num_phys_attrs;
748  ListCell *cur;
749  uint64 processed;
750 
751  if (fe_copy)
752  SendCopyBegin(cstate);
753 
754  if (cstate->rel)
755  tupDesc = RelationGetDescr(cstate->rel);
756  else
757  tupDesc = cstate->queryDesc->tupDesc;
758  num_phys_attrs = tupDesc->natts;
759  cstate->opts.null_print_client = cstate->opts.null_print; /* default */
760 
761  /* We use fe_msgbuf as a per-row buffer regardless of copy_dest */
762  cstate->fe_msgbuf = makeStringInfo();
763 
764  /* Get info about the columns we need to process. */
765  cstate->out_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
766  foreach(cur, cstate->attnumlist)
767  {
768  int attnum = lfirst_int(cur);
769  Oid out_func_oid;
770  bool isvarlena;
771  Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
772 
773  if (cstate->opts.binary)
774  getTypeBinaryOutputInfo(attr->atttypid,
775  &out_func_oid,
776  &isvarlena);
777  else
778  getTypeOutputInfo(attr->atttypid,
779  &out_func_oid,
780  &isvarlena);
781  fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
782  }
783 
784  /*
785  * Create a temporary memory context that we can reset once per row to
786  * recover palloc'd memory. This avoids any problems with leaks inside
787  * datatype output routines, and should be faster than retail pfree's
788  * anyway. (We don't need a whole econtext as CopyFrom does.)
789  */
791  "COPY TO",
793 
794  if (cstate->opts.binary)
795  {
796  /* Generate header for a binary copy */
797  int32 tmp;
798 
799  /* Signature */
800  CopySendData(cstate, BinarySignature, 11);
801  /* Flags field */
802  tmp = 0;
803  CopySendInt32(cstate, tmp);
804  /* No header extension */
805  tmp = 0;
806  CopySendInt32(cstate, tmp);
807  }
808  else
809  {
810  /*
811  * For non-binary copy, we need to convert null_print to file
812  * encoding, because it will be sent directly with CopySendString.
813  */
814  if (cstate->need_transcoding)
816  cstate->opts.null_print_len,
817  cstate->file_encoding);
818 
819  /* if a header has been requested send the line */
820  if (cstate->opts.header_line)
821  {
822  bool hdr_delim = false;
823 
824  foreach(cur, cstate->attnumlist)
825  {
826  int attnum = lfirst_int(cur);
827  char *colname;
828 
829  if (hdr_delim)
830  CopySendChar(cstate, cstate->opts.delim[0]);
831  hdr_delim = true;
832 
833  colname = NameStr(TupleDescAttr(tupDesc, attnum - 1)->attname);
834 
835  if (cstate->opts.csv_mode)
836  CopyAttributeOutCSV(cstate, colname, false);
837  else
838  CopyAttributeOutText(cstate, colname);
839  }
840 
841  CopySendEndOfRow(cstate);
842  }
843  }
844 
845  if (cstate->rel)
846  {
847  TupleTableSlot *slot;
848  TableScanDesc scandesc;
849 
850  scandesc = table_beginscan(cstate->rel, GetActiveSnapshot(), 0, NULL);
851  slot = table_slot_create(cstate->rel, NULL);
852 
853  processed = 0;
854  while (table_scan_getnextslot(scandesc, ForwardScanDirection, slot))
855  {
857 
858  /* Deconstruct the tuple ... */
859  slot_getallattrs(slot);
860 
861  /* Format and send the data */
862  CopyOneRowTo(cstate, slot);
863 
864  /*
865  * Increment the number of processed tuples, and report the
866  * progress.
867  */
869  ++processed);
870  }
871 
873  table_endscan(scandesc);
874  }
875  else
876  {
877  /* run the plan --- the dest receiver will send tuples */
878  ExecutorRun(cstate->queryDesc, ForwardScanDirection, 0, true);
879  processed = ((DR_copy *) cstate->queryDesc->dest)->processed;
880  }
881 
882  if (cstate->opts.binary)
883  {
884  /* Generate trailer for a binary copy */
885  CopySendInt16(cstate, -1);
886  /* Need to flush out the trailer */
887  CopySendEndOfRow(cstate);
888  }
889 
891 
892  if (fe_copy)
893  SendCopyEnd(cstate);
894 
895  return processed;
896 }
signed int int32
Definition: c.h:494
static void SendCopyBegin(CopyToState cstate)
Definition: copyto.c:133
static void SendCopyEnd(CopyToState cstate)
Definition: copyto.c:150
static const char BinarySignature[11]
Definition: copyto.c:106
void ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, uint64 count, bool execute_once)
Definition: execMain.c:297
void ExecDropSingleTupleTableSlot(TupleTableSlot *slot)
Definition: execTuples.c:1341
void fmgr_info(Oid functionId, FmgrInfo *finfo)
Definition: fmgr.c:127
void getTypeBinaryOutputInfo(Oid type, Oid *typSend, bool *typIsVarlena)
Definition: lsyscache.c:2951
void getTypeOutputInfo(Oid type, Oid *typOutput, bool *typIsVarlena)
Definition: lsyscache.c:2885
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:454
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
NameData attname
Definition: pg_attribute.h:41
unsigned int Oid
Definition: postgres_ext.h:31
@ ForwardScanDirection
Definition: sdir.h:28
StringInfo makeStringInfo(void)
Definition: stringinfo.c:41
int null_print_len
Definition: copy.h:67
CopyHeaderChoice header_line
Definition: copy.h:65
DestReceiver * dest
Definition: execdesc.h:41
TupleTableSlot * table_slot_create(Relation relation, List **reglist)
Definition: tableam.c:91
static TableScanDesc table_beginscan(Relation rel, Snapshot snapshot, int nkeys, struct ScanKeyData *key)
Definition: tableam.h:880
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:991
static bool table_scan_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *slot)
Definition: tableam.h:1027

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, attname, attnum, CopyToStateData::attnumlist, CopyFormatOptions::binary, BinarySignature, CHECK_FOR_INTERRUPTS, CopyAttributeOutCSV(), CopyAttributeOutText(), CopyOneRowTo(), CopySendChar(), CopySendData(), CopySendEndOfRow(), CopySendInt16(), CopySendInt32(), CopyFormatOptions::csv_mode, cur, CurrentMemoryContext, CopyToStateData::data_dest_cb, CopyFormatOptions::delim, QueryDesc::dest, DestRemote, ExecDropSingleTupleTableSlot(), ExecutorRun(), CopyToStateData::fe_msgbuf, CopyToStateData::file_encoding, CopyToStateData::filename, fmgr_info(), ForwardScanDirection, GetActiveSnapshot(), getTypeBinaryOutputInfo(), getTypeOutputInfo(), CopyFormatOptions::header_line, lfirst_int, makeStringInfo(), MemoryContextDelete(), NameStr, TupleDescData::natts, CopyToStateData::need_transcoding, CopyFormatOptions::null_print, CopyFormatOptions::null_print_client, CopyFormatOptions::null_print_len, CopyToStateData::opts, CopyToStateData::out_functions, palloc(), pg_server_to_any(), pgstat_progress_update_param(), PROGRESS_COPY_TUPLES_PROCESSED, CopyToStateData::queryDesc, CopyToStateData::rel, RelationGetDescr, CopyToStateData::rowcontext, SendCopyBegin(), SendCopyEnd(), slot_getallattrs(), table_beginscan(), table_endscan(), table_scan_getnextslot(), table_slot_create(), QueryDesc::tupDesc, TupleDescAttr, and whereToSendOutput.

Referenced by DoCopy(), and test_copy_to_callback().

◆ EndCopy()

static void EndCopy ( CopyToState  cstate)
static

Definition at line 314 of file copyto.c.

315 {
316  if (cstate->is_program)
317  {
318  ClosePipeToProgram(cstate);
319  }
320  else
321  {
322  if (cstate->filename != NULL && FreeFile(cstate->copy_file))
323  ereport(ERROR,
325  errmsg("could not close file \"%s\": %m",
326  cstate->filename)));
327  }
328 
330 
332  pfree(cstate);
333 }
void pgstat_progress_end_command(void)
int FreeFile(FILE *file)
Definition: fd.c:2781

References ClosePipeToProgram(), CopyToStateData::copy_file, CopyToStateData::copycontext, ereport, errcode_for_file_access(), errmsg(), ERROR, CopyToStateData::filename, FreeFile(), CopyToStateData::is_program, MemoryContextDelete(), pfree(), and pgstat_progress_end_command().

Referenced by EndCopyTo().

◆ EndCopyTo()

void EndCopyTo ( CopyToState  cstate)

Definition at line 721 of file copyto.c.

722 {
723  if (cstate->queryDesc != NULL)
724  {
725  /* Close down the query and free resources. */
726  ExecutorFinish(cstate->queryDesc);
727  ExecutorEnd(cstate->queryDesc);
728  FreeQueryDesc(cstate->queryDesc);
730  }
731 
732  /* Clean up storage */
733  EndCopy(cstate);
734 }
static void EndCopy(CopyToState cstate)
Definition: copyto.c:314
void ExecutorEnd(QueryDesc *queryDesc)
Definition: execMain.c:467
void ExecutorFinish(QueryDesc *queryDesc)
Definition: execMain.c:407
void FreeQueryDesc(QueryDesc *qdesc)
Definition: pquery.c:105
void PopActiveSnapshot(void)
Definition: snapmgr.c:743

References EndCopy(), ExecutorEnd(), ExecutorFinish(), FreeQueryDesc(), PopActiveSnapshot(), and CopyToStateData::queryDesc.

Referenced by DoCopy(), and test_copy_to_callback().

◆ SendCopyBegin()

static void SendCopyBegin ( CopyToState  cstate)
static

Definition at line 133 of file copyto.c.

134 {
136  int natts = list_length(cstate->attnumlist);
137  int16 format = (cstate->opts.binary ? 1 : 0);
138  int i;
139 
141  pq_sendbyte(&buf, format); /* overall format */
142  pq_sendint16(&buf, natts);
143  for (i = 0; i < natts; i++)
144  pq_sendint16(&buf, format); /* per-column formats */
145  pq_endmessage(&buf);
146  cstate->copy_dest = COPY_FRONTEND;
147 }
signed short int16
Definition: c.h:493
int i
Definition: isn.c:73
static char format
void pq_endmessage(StringInfo buf)
Definition: pqformat.c:296
void pq_beginmessage(StringInfo buf, char msgtype)
Definition: pqformat.c:88
static void pq_sendbyte(StringInfo buf, uint8 byt)
Definition: pqformat.h:160
static void pq_sendint16(StringInfo buf, uint16 i)
Definition: pqformat.h:136
#define PqMsg_CopyOutResponse
Definition: protocol.h:46

References CopyToStateData::attnumlist, CopyFormatOptions::binary, buf, CopyToStateData::copy_dest, COPY_FRONTEND, format, i, list_length(), CopyToStateData::opts, pq_beginmessage(), pq_endmessage(), pq_sendbyte(), pq_sendint16(), and PqMsg_CopyOutResponse.

Referenced by DoCopyTo().

◆ SendCopyEnd()

static void SendCopyEnd ( CopyToState  cstate)
static

Definition at line 150 of file copyto.c.

151 {
152  /* Shouldn't have any unsent data */
153  Assert(cstate->fe_msgbuf->len == 0);
154  /* Send Copy Done message */
156 }
void pq_putemptymessage(char msgtype)
Definition: pqformat.c:388
#define PqMsg_CopyDone
Definition: protocol.h:64

References Assert, CopyToStateData::fe_msgbuf, StringInfoData::len, pq_putemptymessage(), and PqMsg_CopyDone.

Referenced by DoCopyTo().

Variable Documentation

◆ BinarySignature

const char BinarySignature[11] = "PGCOPY\n\377\r\n\0"
static

Definition at line 106 of file copyto.c.

Referenced by DoCopyTo().