Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * copy.c
4 : * Implements the COPY utility command
5 : *
6 : * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
7 : * Portions Copyright (c) 1994, Regents of the University of California
8 : *
9 : *
10 : * IDENTIFICATION
11 : * src/backend/commands/copy.c
12 : *
13 : *-------------------------------------------------------------------------
14 : */
15 : #include "postgres.h"
16 :
17 : #include <ctype.h>
18 : #include <unistd.h>
19 : #include <sys/stat.h>
20 :
21 : #include "access/heapam.h"
22 : #include "access/htup_details.h"
23 : #include "access/sysattr.h"
24 : #include "access/tableam.h"
25 : #include "access/xact.h"
26 : #include "access/xlog.h"
27 : #include "catalog/dependency.h"
28 : #include "catalog/pg_authid.h"
29 : #include "catalog/pg_type.h"
30 : #include "commands/copy.h"
31 : #include "commands/defrem.h"
32 : #include "commands/trigger.h"
33 : #include "executor/execPartition.h"
34 : #include "executor/executor.h"
35 : #include "executor/nodeModifyTable.h"
36 : #include "executor/tuptable.h"
37 : #include "foreign/fdwapi.h"
38 : #include "libpq/libpq.h"
39 : #include "libpq/pqformat.h"
40 : #include "mb/pg_wchar.h"
41 : #include "miscadmin.h"
42 : #include "nodes/makefuncs.h"
43 : #include "optimizer/optimizer.h"
44 : #include "parser/parse_coerce.h"
45 : #include "parser/parse_collate.h"
46 : #include "parser/parse_expr.h"
47 : #include "parser/parse_relation.h"
48 : #include "port/pg_bswap.h"
49 : #include "rewrite/rewriteHandler.h"
50 : #include "storage/fd.h"
51 : #include "tcop/tcopprot.h"
52 : #include "utils/acl.h"
53 : #include "utils/builtins.h"
54 : #include "utils/lsyscache.h"
55 : #include "utils/memutils.h"
56 : #include "utils/partcache.h"
57 : #include "utils/portal.h"
58 : #include "utils/rel.h"
59 : #include "utils/rls.h"
60 : #include "utils/snapmgr.h"
61 :
62 : #define ISOCTAL(c) (((c) >= '0') && ((c) <= '7'))
63 : #define OCTVALUE(c) ((c) - '0')
64 :
65 : /*
66 : * Represents the different source/dest cases we need to worry about at
67 : * the bottom level
68 : */
69 : typedef enum CopyDest
70 : {
71 : COPY_FILE, /* to/from file (or a piped program) */
72 : COPY_OLD_FE, /* to/from frontend (2.0 protocol) */
73 : COPY_NEW_FE, /* to/from frontend (3.0 protocol) */
74 : COPY_CALLBACK /* to/from callback function */
75 : } CopyDest;
76 :
77 : /*
78 : * Represents the end-of-line terminator type of the input
79 : */
80 : typedef enum EolType
81 : {
82 : EOL_UNKNOWN,
83 : EOL_NL,
84 : EOL_CR,
85 : EOL_CRNL
86 : } EolType;
87 :
88 : /*
89 : * Represents the heap insert method to be used during COPY FROM.
90 : */
91 : typedef enum CopyInsertMethod
92 : {
93 : CIM_SINGLE, /* use table_tuple_insert or fdw routine */
94 : CIM_MULTI, /* always use table_multi_insert */
95 : CIM_MULTI_CONDITIONAL /* use table_multi_insert only if valid */
96 : } CopyInsertMethod;
97 :
98 : /*
99 : * This struct contains all the state variables used throughout a COPY
100 : * operation. For simplicity, we use the same struct for all variants of COPY,
101 : * even though some fields are used in only some cases.
102 : *
103 : * Multi-byte encodings: all supported client-side encodings encode multi-byte
104 : * characters by having the first byte's high bit set. Subsequent bytes of the
105 : * character can have the high bit not set. When scanning data in such an
106 : * encoding to look for a match to a single-byte (ie ASCII) character, we must
107 : * use the full pg_encoding_mblen() machinery to skip over multibyte
108 : * characters, else we might find a false match to a trailing byte. In
109 : * supported server encodings, there is no possibility of a false match, and
110 : * it's faster to make useless comparisons to trailing bytes than it is to
111 : * invoke pg_encoding_mblen() to skip over them. encoding_embeds_ascii is true
112 : * when we have to do it the hard way.
113 : */
114 : typedef struct CopyStateData
115 : {
116 : /* low-level state data */
117 : CopyDest copy_dest; /* type of copy source/destination */
118 : FILE *copy_file; /* used if copy_dest == COPY_FILE */
119 : StringInfo fe_msgbuf; /* used for all dests during COPY TO, only for
120 : * dest == COPY_NEW_FE in COPY FROM */
121 : bool is_copy_from; /* COPY TO, or COPY FROM? */
122 : bool reached_eof; /* true if we read to end of copy data (not
123 : * all copy_dest types maintain this) */
124 : EolType eol_type; /* EOL type of input */
125 : int file_encoding; /* file or remote side's character encoding */
126 : bool need_transcoding; /* file encoding diff from server? */
127 : bool encoding_embeds_ascii; /* ASCII can be non-first byte? */
128 :
129 : /* parameters from the COPY command */
130 : Relation rel; /* relation to copy to or from */
131 : QueryDesc *queryDesc; /* executable query to copy from */
132 : List *attnumlist; /* integer list of attnums to copy */
133 : char *filename; /* filename, or NULL for STDIN/STDOUT */
134 : bool is_program; /* is 'filename' a program to popen? */
135 : copy_data_source_cb data_source_cb; /* function for reading data */
136 : bool binary; /* binary format? */
137 : bool freeze; /* freeze rows on loading? */
138 : bool csv_mode; /* Comma Separated Value format? */
139 : bool header_line; /* CSV header line? */
140 : char *null_print; /* NULL marker string (server encoding!) */
141 : int null_print_len; /* length of same */
142 : char *null_print_client; /* same converted to file encoding */
143 : char *delim; /* column delimiter (must be 1 byte) */
144 : char *quote; /* CSV quote char (must be 1 byte) */
145 : char *escape; /* CSV escape char (must be 1 byte) */
146 : List *force_quote; /* list of column names */
147 : bool force_quote_all; /* FORCE_QUOTE *? */
148 : bool *force_quote_flags; /* per-column CSV FQ flags */
149 : List *force_notnull; /* list of column names */
150 : bool *force_notnull_flags; /* per-column CSV FNN flags */
151 : List *force_null; /* list of column names */
152 : bool *force_null_flags; /* per-column CSV FN flags */
153 : bool convert_selectively; /* do selective binary conversion? */
154 : List *convert_select; /* list of column names (can be NIL) */
155 : bool *convert_select_flags; /* per-column CSV/TEXT CS flags */
156 : Node *whereClause; /* WHERE condition (or NULL) */
157 :
158 : /* these are just for error messages, see CopyFromErrorCallback */
159 : const char *cur_relname; /* table name for error messages */
160 : uint64 cur_lineno; /* line number for error messages */
161 : const char *cur_attname; /* current att for error messages */
162 : const char *cur_attval; /* current att value for error messages */
163 :
164 : /*
165 : * Working state for COPY TO/FROM
166 : */
167 : MemoryContext copycontext; /* per-copy execution context */
168 :
169 : /*
170 : * Working state for COPY TO
171 : */
172 : FmgrInfo *out_functions; /* lookup info for output functions */
173 : MemoryContext rowcontext; /* per-row evaluation context */
174 :
175 : /*
176 : * Working state for COPY FROM
177 : */
178 : AttrNumber num_defaults;
179 : FmgrInfo *in_functions; /* array of input functions for each attrs */
180 : Oid *typioparams; /* array of element types for in_functions */
181 : int *defmap; /* array of default att numbers */
182 : ExprState **defexprs; /* array of default att expressions */
183 : bool volatile_defexprs; /* is any of defexprs volatile? */
184 : List *range_table;
185 : ExprState *qualexpr;
186 :
187 : TransitionCaptureState *transition_capture;
188 :
189 : /*
190 : * These variables are used to reduce overhead in textual COPY FROM.
191 : *
192 : * attribute_buf holds the separated, de-escaped text for each field of
193 : * the current line. The CopyReadAttributes functions return arrays of
194 : * pointers into this buffer. We avoid palloc/pfree overhead by re-using
195 : * the buffer on each cycle.
196 : */
197 : StringInfoData attribute_buf;
198 :
199 : /* field raw data pointers found by COPY FROM */
200 :
201 : int max_fields;
202 : char **raw_fields;
203 :
204 : /*
205 : * Similarly, line_buf holds the whole input line being processed. The
206 : * input cycle is first to read the whole line into line_buf, convert it
207 : * to server encoding there, and then extract the individual attribute
208 : * fields into attribute_buf. line_buf is preserved unmodified so that we
209 : * can display it in error messages if appropriate.
210 : */
211 : StringInfoData line_buf;
212 : bool line_buf_converted; /* converted to server encoding? */
213 : bool line_buf_valid; /* contains the row being processed? */
214 :
215 : /*
216 : * Finally, raw_buf holds raw data read from the data source (file or
217 : * client connection). CopyReadLine parses this data sufficiently to
218 : * locate line boundaries, then transfers the data to line_buf and
219 : * converts it. Note: we guarantee that there is a \0 at
220 : * raw_buf[raw_buf_len].
221 : */
222 : #define RAW_BUF_SIZE 65536 /* we palloc RAW_BUF_SIZE+1 bytes */
223 : char *raw_buf;
224 : int raw_buf_index; /* next byte to process */
225 : int raw_buf_len; /* total # of bytes stored */
226 : } CopyStateData;
227 :
228 : /* DestReceiver for COPY (query) TO */
229 : typedef struct
230 : {
231 : DestReceiver pub; /* publicly-known function pointers */
232 : CopyState cstate; /* CopyStateData for the command */
233 : uint64 processed; /* # of tuples processed */
234 : } DR_copy;
235 :
236 :
237 : /*
238 : * No more than this many tuples per CopyMultiInsertBuffer
239 : *
240 : * Caution: Don't make this too big, as we could end up with this many
241 : * CopyMultiInsertBuffer items stored in CopyMultiInsertInfo's
242 : * multiInsertBuffers list. Increasing this can cause quadratic growth in
243 : * memory requirements during copies into partitioned tables with a large
244 : * number of partitions.
245 : */
246 : #define MAX_BUFFERED_TUPLES 1000
247 :
248 : /*
249 : * Flush buffers if there are >= this many bytes, as counted by the input
250 : * size, of tuples stored.
251 : */
252 : #define MAX_BUFFERED_BYTES 65535
253 :
254 : /* Trim the list of buffers back down to this number after flushing */
255 : #define MAX_PARTITION_BUFFERS 32
256 :
257 : /* Stores multi-insert data related to a single relation in CopyFrom. */
258 : typedef struct CopyMultiInsertBuffer
259 : {
260 : TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */
261 : ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */
262 : BulkInsertState bistate; /* BulkInsertState for this rel */
263 : int nused; /* number of 'slots' containing tuples */
264 : uint64 linenos[MAX_BUFFERED_TUPLES]; /* Line # of tuple in copy
265 : * stream */
266 : } CopyMultiInsertBuffer;
267 :
268 : /*
269 : * Stores one or many CopyMultiInsertBuffers and details about the size and
270 : * number of tuples which are stored in them. This allows multiple buffers to
271 : * exist at once when COPYing into a partitioned table.
272 : */
273 : typedef struct CopyMultiInsertInfo
274 : {
275 : List *multiInsertBuffers; /* List of tracked CopyMultiInsertBuffers */
276 : int bufferedTuples; /* number of tuples buffered over all buffers */
277 : int bufferedBytes; /* number of bytes from all buffered tuples */
278 : CopyState cstate; /* Copy state for this CopyMultiInsertInfo */
279 : EState *estate; /* Executor state used for COPY */
280 : CommandId mycid; /* Command Id used for COPY */
281 : int ti_options; /* table insert options */
282 : } CopyMultiInsertInfo;
283 :
284 :
285 : /*
286 : * These macros centralize code used to process line_buf and raw_buf buffers.
287 : * They are macros because they often do continue/break control and to avoid
288 : * function call overhead in tight COPY loops.
289 : *
290 : * We must use "if (1)" because the usual "do {...} while(0)" wrapper would
291 : * prevent the continue/break processing from working. We end the "if (1)"
292 : * with "else ((void) 0)" to ensure the "if" does not unintentionally match
293 : * any "else" in the calling code, and to avoid any compiler warnings about
294 : * empty statements. See http://www.cit.gu.edu.au/~anthony/info/C/C.macros.
295 : */
296 :
297 : /*
298 : * This keeps the character read at the top of the loop in the buffer
299 : * even if there is more than one read-ahead.
300 : */
301 : #define IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(extralen) \
302 : if (1) \
303 : { \
304 : if (raw_buf_ptr + (extralen) >= copy_buf_len && !hit_eof) \
305 : { \
306 : raw_buf_ptr = prev_raw_ptr; /* undo fetch */ \
307 : need_data = true; \
308 : continue; \
309 : } \
310 : } else ((void) 0)
311 :
312 : /* This consumes the remainder of the buffer and breaks */
313 : #define IF_NEED_REFILL_AND_EOF_BREAK(extralen) \
314 : if (1) \
315 : { \
316 : if (raw_buf_ptr + (extralen) >= copy_buf_len && hit_eof) \
317 : { \
318 : if (extralen) \
319 : raw_buf_ptr = copy_buf_len; /* consume the partial character */ \
320 : /* backslash just before EOF, treat as data char */ \
321 : result = true; \
322 : break; \
323 : } \
324 : } else ((void) 0)
325 :
326 : /*
327 : * Transfer any approved data to line_buf; must do this to be sure
328 : * there is some room in raw_buf.
329 : */
330 : #define REFILL_LINEBUF \
331 : if (1) \
332 : { \
333 : if (raw_buf_ptr > cstate->raw_buf_index) \
334 : { \
335 : appendBinaryStringInfo(&cstate->line_buf, \
336 : cstate->raw_buf + cstate->raw_buf_index, \
337 : raw_buf_ptr - cstate->raw_buf_index); \
338 : cstate->raw_buf_index = raw_buf_ptr; \
339 : } \
340 : } else ((void) 0)
341 :
342 : /* Undo any read-ahead and jump out of the block. */
343 : #define NO_END_OF_COPY_GOTO \
344 : if (1) \
345 : { \
346 : raw_buf_ptr = prev_raw_ptr + 1; \
347 : goto not_end_of_copy; \
348 : } else ((void) 0)
349 :
350 : static const char BinarySignature[11] = "PGCOPY\n\377\r\n\0";
351 :
352 :
353 : /* non-export function prototypes */
354 : static CopyState BeginCopy(ParseState *pstate, bool is_from, Relation rel,
355 : RawStmt *raw_query, Oid queryRelId, List *attnamelist,
356 : List *options);
357 : static void EndCopy(CopyState cstate);
358 : static void ClosePipeToProgram(CopyState cstate);
359 : static CopyState BeginCopyTo(ParseState *pstate, Relation rel, RawStmt *query,
360 : Oid queryRelId, const char *filename, bool is_program,
361 : List *attnamelist, List *options);
362 : static void EndCopyTo(CopyState cstate);
363 : static uint64 DoCopyTo(CopyState cstate);
364 : static uint64 CopyTo(CopyState cstate);
365 : static void CopyOneRowTo(CopyState cstate, TupleTableSlot *slot);
366 : static bool CopyReadLine(CopyState cstate);
367 : static bool CopyReadLineText(CopyState cstate);
368 : static int CopyReadAttributesText(CopyState cstate);
369 : static int CopyReadAttributesCSV(CopyState cstate);
370 : static Datum CopyReadBinaryAttribute(CopyState cstate,
371 : int column_no, FmgrInfo *flinfo,
372 : Oid typioparam, int32 typmod,
373 : bool *isnull);
374 : static void CopyAttributeOutText(CopyState cstate, char *string);
375 : static void CopyAttributeOutCSV(CopyState cstate, char *string,
376 : bool use_quote, bool single_attr);
377 : static List *CopyGetAttnums(TupleDesc tupDesc, Relation rel,
378 : List *attnamelist);
379 : static char *limit_printout_length(const char *str);
380 :
381 : /* Low-level communications functions */
382 : static void SendCopyBegin(CopyState cstate);
383 : static void ReceiveCopyBegin(CopyState cstate);
384 : static void SendCopyEnd(CopyState cstate);
385 : static void CopySendData(CopyState cstate, const void *databuf, int datasize);
386 : static void CopySendString(CopyState cstate, const char *str);
387 : static void CopySendChar(CopyState cstate, char c);
388 : static void CopySendEndOfRow(CopyState cstate);
389 : static int CopyGetData(CopyState cstate, void *databuf,
390 : int minread, int maxread);
391 : static void CopySendInt32(CopyState cstate, int32 val);
392 : static bool CopyGetInt32(CopyState cstate, int32 *val);
393 : static void CopySendInt16(CopyState cstate, int16 val);
394 : static bool CopyGetInt16(CopyState cstate, int16 *val);
395 :
396 :
397 : /*
398 : * Send copy start/stop messages for frontend copies. These have changed
399 : * in past protocol redesigns.
400 : */
401 : static void
402 2926 : SendCopyBegin(CopyState cstate)
403 : {
404 2926 : if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 3)
405 : {
406 : /* new way */
407 : StringInfoData buf;
408 2926 : int natts = list_length(cstate->attnumlist);
409 2926 : int16 format = (cstate->binary ? 1 : 0);
410 : int i;
411 :
412 2926 : pq_beginmessage(&buf, 'H');
413 2926 : pq_sendbyte(&buf, format); /* overall format */
414 2926 : pq_sendint16(&buf, natts);
415 13832 : for (i = 0; i < natts; i++)
416 10906 : pq_sendint16(&buf, format); /* per-column formats */
417 2926 : pq_endmessage(&buf);
418 2926 : cstate->copy_dest = COPY_NEW_FE;
419 : }
420 : else
421 : {
422 : /* old way */
423 0 : if (cstate->binary)
424 0 : ereport(ERROR,
425 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
426 : errmsg("COPY BINARY is not supported to stdout or from stdin")));
427 0 : pq_putemptymessage('H');
428 : /* grottiness needed for old COPY OUT protocol */
429 0 : pq_startcopyout();
430 0 : cstate->copy_dest = COPY_OLD_FE;
431 : }
432 2926 : }
433 :
434 : static void
435 472 : ReceiveCopyBegin(CopyState cstate)
436 : {
437 472 : if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 3)
438 : {
439 : /* new way */
440 : StringInfoData buf;
441 472 : int natts = list_length(cstate->attnumlist);
442 472 : int16 format = (cstate->binary ? 1 : 0);
443 : int i;
444 :
445 472 : pq_beginmessage(&buf, 'G');
446 472 : pq_sendbyte(&buf, format); /* overall format */
447 472 : pq_sendint16(&buf, natts);
448 1502 : for (i = 0; i < natts; i++)
449 1030 : pq_sendint16(&buf, format); /* per-column formats */
450 472 : pq_endmessage(&buf);
451 472 : cstate->copy_dest = COPY_NEW_FE;
452 472 : cstate->fe_msgbuf = makeStringInfo();
453 : }
454 : else
455 : {
456 : /* old way */
457 0 : if (cstate->binary)
458 0 : ereport(ERROR,
459 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
460 : errmsg("COPY BINARY is not supported to stdout or from stdin")));
461 0 : pq_putemptymessage('G');
462 : /* any error in old protocol will make us lose sync */
463 0 : pq_startmsgread();
464 0 : cstate->copy_dest = COPY_OLD_FE;
465 : }
466 : /* We *must* flush here to ensure FE knows it can send. */
467 472 : pq_flush();
468 472 : }
469 :
470 : static void
471 2926 : SendCopyEnd(CopyState cstate)
472 : {
473 2926 : if (cstate->copy_dest == COPY_NEW_FE)
474 : {
475 : /* Shouldn't have any unsent data */
476 : Assert(cstate->fe_msgbuf->len == 0);
477 : /* Send Copy Done message */
478 2926 : pq_putemptymessage('c');
479 : }
480 : else
481 : {
482 0 : CopySendData(cstate, "\\.", 2);
483 : /* Need to flush out the trailer (this also appends a newline) */
484 0 : CopySendEndOfRow(cstate);
485 0 : pq_endcopyout(false);
486 : }
487 2926 : }
488 :
489 : /*----------
490 : * CopySendData sends output data to the destination (file or frontend)
491 : * CopySendString does the same for null-terminated strings
492 : * CopySendChar does the same for single characters
493 : * CopySendEndOfRow does the appropriate thing at end of each data row
494 : * (data is not actually flushed except by CopySendEndOfRow)
495 : *
496 : * NB: no data conversion is applied by these functions
497 : *----------
498 : */
499 : static void
500 4558652 : CopySendData(CopyState cstate, const void *databuf, int datasize)
501 : {
502 4558652 : appendBinaryStringInfo(cstate->fe_msgbuf, databuf, datasize);
503 4558652 : }
504 :
505 : static void
506 282234 : CopySendString(CopyState cstate, const char *str)
507 : {
508 282234 : appendBinaryStringInfo(cstate->fe_msgbuf, str, strlen(str));
509 282234 : }
510 :
511 : static void
512 4848606 : CopySendChar(CopyState cstate, char c)
513 : {
514 4848606 : appendStringInfoCharMacro(cstate->fe_msgbuf, c);
515 4848606 : }
516 :
517 : static void
518 1473248 : CopySendEndOfRow(CopyState cstate)
519 : {
520 1473248 : StringInfo fe_msgbuf = cstate->fe_msgbuf;
521 :
522 1473248 : switch (cstate->copy_dest)
523 : {
524 12134 : case COPY_FILE:
525 12134 : if (!cstate->binary)
526 : {
527 : /* Default line termination depends on platform */
528 : #ifndef WIN32
529 12118 : CopySendChar(cstate, '\n');
530 : #else
531 : CopySendString(cstate, "\r\n");
532 : #endif
533 : }
534 :
535 12134 : if (fwrite(fe_msgbuf->data, fe_msgbuf->len, 1,
536 12134 : cstate->copy_file) != 1 ||
537 12134 : ferror(cstate->copy_file))
538 : {
539 0 : if (cstate->is_program)
540 : {
541 0 : if (errno == EPIPE)
542 : {
543 : /*
544 : * The pipe will be closed automatically on error at
545 : * the end of transaction, but we might get a better
546 : * error message from the subprocess' exit code than
547 : * just "Broken Pipe"
548 : */
549 0 : ClosePipeToProgram(cstate);
550 :
551 : /*
552 : * If ClosePipeToProgram() didn't throw an error, the
553 : * program terminated normally, but closed the pipe
554 : * first. Restore errno, and throw an error.
555 : */
556 0 : errno = EPIPE;
557 : }
558 0 : ereport(ERROR,
559 : (errcode_for_file_access(),
560 : errmsg("could not write to COPY program: %m")));
561 : }
562 : else
563 0 : ereport(ERROR,
564 : (errcode_for_file_access(),
565 : errmsg("could not write to COPY file: %m")));
566 : }
567 12134 : break;
568 0 : case COPY_OLD_FE:
569 : /* The FE/BE protocol uses \n as newline for all platforms */
570 0 : if (!cstate->binary)
571 0 : CopySendChar(cstate, '\n');
572 :
573 0 : if (pq_putbytes(fe_msgbuf->data, fe_msgbuf->len))
574 : {
575 : /* no hope of recovering connection sync, so FATAL */
576 0 : ereport(FATAL,
577 : (errcode(ERRCODE_CONNECTION_FAILURE),
578 : errmsg("connection lost during COPY to stdout")));
579 : }
580 0 : break;
581 1461114 : case COPY_NEW_FE:
582 : /* The FE/BE protocol uses \n as newline for all platforms */
583 1461114 : if (!cstate->binary)
584 1461114 : CopySendChar(cstate, '\n');
585 :
586 : /* Dump the accumulated row as one CopyData message */
587 1461114 : (void) pq_putmessage('d', fe_msgbuf->data, fe_msgbuf->len);
588 1461114 : break;
589 0 : case COPY_CALLBACK:
590 : Assert(false); /* Not yet supported. */
591 0 : break;
592 : }
593 :
594 1473248 : resetStringInfo(fe_msgbuf);
595 1473248 : }
596 :
597 : /*
598 : * CopyGetData reads data from the source (file or frontend)
599 : *
600 : * We attempt to read at least minread, and at most maxread, bytes from
601 : * the source. The actual number of bytes read is returned; if this is
602 : * less than minread, EOF was detected.
603 : *
604 : * Note: when copying from the frontend, we expect a proper EOF mark per
605 : * protocol; if the frontend simply drops the connection, we raise error.
606 : * It seems unwise to allow the COPY IN to complete normally in that case.
607 : *
608 : * NB: no data conversion is applied here.
609 : */
610 : static int
611 477888 : CopyGetData(CopyState cstate, void *databuf, int minread, int maxread)
612 : {
613 477888 : int bytesread = 0;
614 :
615 477888 : switch (cstate->copy_dest)
616 : {
617 1404 : case COPY_FILE:
618 1404 : bytesread = fread(databuf, 1, maxread, cstate->copy_file);
619 1404 : if (ferror(cstate->copy_file))
620 0 : ereport(ERROR,
621 : (errcode_for_file_access(),
622 : errmsg("could not read from COPY file: %m")));
623 1404 : if (bytesread == 0)
624 542 : cstate->reached_eof = true;
625 1404 : break;
626 0 : case COPY_OLD_FE:
627 :
628 : /*
629 : * We cannot read more than minread bytes (which in practice is 1)
630 : * because old protocol doesn't have any clear way of separating
631 : * the COPY stream from following data. This is slow, but not any
632 : * slower than the code path was originally, and we don't care
633 : * much anymore about the performance of old protocol.
634 : */
635 0 : if (pq_getbytes((char *) databuf, minread))
636 : {
637 : /* Only a \. terminator is legal EOF in old protocol */
638 0 : ereport(ERROR,
639 : (errcode(ERRCODE_CONNECTION_FAILURE),
640 : errmsg("unexpected EOF on client connection with an open transaction")));
641 : }
642 0 : bytesread = minread;
643 0 : break;
644 474140 : case COPY_NEW_FE:
645 947480 : while (maxread > 0 && bytesread < minread && !cstate->reached_eof)
646 : {
647 : int avail;
648 :
649 947084 : while (cstate->fe_msgbuf->cursor >= cstate->fe_msgbuf->len)
650 : {
651 : /* Try to receive another message */
652 : int mtype;
653 :
654 473744 : readmessage:
655 473744 : HOLD_CANCEL_INTERRUPTS();
656 473744 : pq_startmsgread();
657 473744 : mtype = pq_getbyte();
658 473744 : if (mtype == EOF)
659 0 : ereport(ERROR,
660 : (errcode(ERRCODE_CONNECTION_FAILURE),
661 : errmsg("unexpected EOF on client connection with an open transaction")));
662 473744 : if (pq_getmessage(cstate->fe_msgbuf, 0))
663 0 : ereport(ERROR,
664 : (errcode(ERRCODE_CONNECTION_FAILURE),
665 : errmsg("unexpected EOF on client connection with an open transaction")));
666 473744 : RESUME_CANCEL_INTERRUPTS();
667 : switch (mtype)
668 : {
669 473340 : case 'd': /* CopyData */
670 473340 : break;
671 404 : case 'c': /* CopyDone */
672 : /* COPY IN correctly terminated by frontend */
673 404 : cstate->reached_eof = true;
674 404 : return bytesread;
675 0 : case 'f': /* CopyFail */
676 0 : ereport(ERROR,
677 : (errcode(ERRCODE_QUERY_CANCELED),
678 : errmsg("COPY from stdin failed: %s",
679 : pq_getmsgstring(cstate->fe_msgbuf))));
680 : break;
681 0 : case 'H': /* Flush */
682 : case 'S': /* Sync */
683 :
684 : /*
685 : * Ignore Flush/Sync for the convenience of client
686 : * libraries (such as libpq) that may send those
687 : * without noticing that the command they just
688 : * sent was COPY.
689 : */
690 0 : goto readmessage;
691 0 : default:
692 0 : ereport(ERROR,
693 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
694 : errmsg("unexpected message type 0x%02X during COPY from stdin",
695 : mtype)));
696 : break;
697 : }
698 : }
699 473340 : avail = cstate->fe_msgbuf->len - cstate->fe_msgbuf->cursor;
700 473340 : if (avail > maxread)
701 0 : avail = maxread;
702 473340 : pq_copymsgbytes(cstate->fe_msgbuf, databuf, avail);
703 473340 : databuf = (void *) ((char *) databuf + avail);
704 473340 : maxread -= avail;
705 473340 : bytesread += avail;
706 : }
707 473736 : break;
708 2344 : case COPY_CALLBACK:
709 2344 : bytesread = cstate->data_source_cb(databuf, minread, maxread);
710 2344 : break;
711 : }
712 :
713 477484 : return bytesread;
714 : }
715 :
716 :
717 : /*
718 : * These functions do apply some data conversion
719 : */
720 :
721 : /*
722 : * CopySendInt32 sends an int32 in network byte order
723 : */
724 : static void
725 92 : CopySendInt32(CopyState cstate, int32 val)
726 : {
727 : uint32 buf;
728 :
729 92 : buf = pg_hton32((uint32) val);
730 92 : CopySendData(cstate, &buf, sizeof(buf));
731 92 : }
732 :
733 : /*
734 : * CopyGetInt32 reads an int32 that appears in network byte order
735 : *
736 : * Returns true if OK, false if EOF
737 : */
738 : static bool
739 92 : CopyGetInt32(CopyState cstate, int32 *val)
740 : {
741 : uint32 buf;
742 :
743 92 : if (CopyGetData(cstate, &buf, sizeof(buf), sizeof(buf)) != sizeof(buf))
744 : {
745 0 : *val = 0; /* suppress compiler warning */
746 0 : return false;
747 : }
748 92 : *val = (int32) pg_ntoh32(buf);
749 92 : return true;
750 : }
751 :
752 : /*
753 : * CopySendInt16 sends an int16 in network byte order
754 : */
755 : static void
756 16 : CopySendInt16(CopyState cstate, int16 val)
757 : {
758 : uint16 buf;
759 :
760 16 : buf = pg_hton16((uint16) val);
761 16 : CopySendData(cstate, &buf, sizeof(buf));
762 16 : }
763 :
764 : /*
765 : * CopyGetInt16 reads an int16 that appears in network byte order
766 : */
767 : static bool
768 16 : CopyGetInt16(CopyState cstate, int16 *val)
769 : {
770 : uint16 buf;
771 :
772 16 : if (CopyGetData(cstate, &buf, sizeof(buf), sizeof(buf)) != sizeof(buf))
773 : {
774 0 : *val = 0; /* suppress compiler warning */
775 0 : return false;
776 : }
777 16 : *val = (int16) pg_ntoh16(buf);
778 16 : return true;
779 : }
780 :
781 :
782 : /*
783 : * CopyLoadRawBuf loads some more data into raw_buf
784 : *
785 : * Returns true if able to obtain at least one more byte, else false.
786 : *
787 : * If raw_buf_index < raw_buf_len, the unprocessed bytes are transferred
788 : * down to the start of the buffer and then we load more data after that.
789 : * This case is used only when a frontend multibyte character crosses a
790 : * bufferload boundary.
791 : */
792 : static bool
793 477708 : CopyLoadRawBuf(CopyState cstate)
794 : {
795 : int nbytes;
796 : int inbytes;
797 :
798 477708 : if (cstate->raw_buf_index < cstate->raw_buf_len)
799 : {
800 : /* Copy down the unprocessed data */
801 0 : nbytes = cstate->raw_buf_len - cstate->raw_buf_index;
802 0 : memmove(cstate->raw_buf, cstate->raw_buf + cstate->raw_buf_index,
803 : nbytes);
804 : }
805 : else
806 477708 : nbytes = 0; /* no data need be saved */
807 :
808 477708 : inbytes = CopyGetData(cstate, cstate->raw_buf + nbytes,
809 : 1, RAW_BUF_SIZE - nbytes);
810 477708 : nbytes += inbytes;
811 477708 : cstate->raw_buf[nbytes] = '\0';
812 477708 : cstate->raw_buf_index = 0;
813 477708 : cstate->raw_buf_len = nbytes;
814 477708 : return (inbytes > 0);
815 : }
816 :
817 :
818 : /*
819 : * DoCopy executes the SQL COPY statement
820 : *
821 : * Either unload or reload contents of table <relation>, depending on <from>.
822 : * (<from> = true means we are inserting into the table.) In the "TO" case
823 : * we also support copying the output of an arbitrary SELECT, INSERT, UPDATE
824 : * or DELETE query.
825 : *
826 : * If <pipe> is false, transfer is between the table and the file named
827 : * <filename>. Otherwise, transfer is between the table and our regular
828 : * input/output stream. The latter could be either stdin/stdout or a
829 : * socket, depending on whether we're running under Postmaster control.
830 : *
831 : * Do not allow a Postgres user without the 'pg_read_server_files' or
832 : * 'pg_write_server_files' role to read from or write to a file.
833 : *
834 : * Do not allow the copy if user doesn't have proper permission to access
835 : * the table or the specifically requested columns.
836 : */
837 : void
838 4146 : DoCopy(ParseState *pstate, const CopyStmt *stmt,
839 : int stmt_location, int stmt_len,
840 : uint64 *processed)
841 : {
842 : CopyState cstate;
843 4146 : bool is_from = stmt->is_from;
844 4146 : bool pipe = (stmt->filename == NULL);
845 : Relation rel;
846 : Oid relid;
847 4146 : RawStmt *query = NULL;
848 4146 : Node *whereClause = NULL;
849 :
850 : /*
851 : * Disallow COPY to/from file or program except to users with the
852 : * appropriate role.
853 : */
854 4146 : if (!pipe)
855 : {
856 532 : if (stmt->is_program)
857 : {
858 0 : if (!is_member_of_role(GetUserId(), DEFAULT_ROLE_EXECUTE_SERVER_PROGRAM))
859 0 : ereport(ERROR,
860 : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
861 : errmsg("must be superuser or a member of the pg_execute_server_program role to COPY to or from an external program"),
862 : errhint("Anyone can COPY to stdout or from stdin. "
863 : "psql's \\copy command also works for anyone.")));
864 : }
865 : else
866 : {
867 532 : if (is_from && !is_member_of_role(GetUserId(), DEFAULT_ROLE_READ_SERVER_FILES))
868 0 : ereport(ERROR,
869 : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
870 : errmsg("must be superuser or a member of the pg_read_server_files role to COPY from a file"),
871 : errhint("Anyone can COPY to stdout or from stdin. "
872 : "psql's \\copy command also works for anyone.")));
873 :
874 532 : if (!is_from && !is_member_of_role(GetUserId(), DEFAULT_ROLE_WRITE_SERVER_FILES))
875 0 : ereport(ERROR,
876 : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
877 : errmsg("must be superuser or a member of the pg_write_server_files role to COPY to a file"),
878 : errhint("Anyone can COPY to stdout or from stdin. "
879 : "psql's \\copy command also works for anyone.")));
880 : }
881 : }
882 :
883 4146 : if (stmt->relation)
884 : {
885 3956 : LOCKMODE lockmode = is_from ? RowExclusiveLock : AccessShareLock;
886 : ParseNamespaceItem *nsitem;
887 : RangeTblEntry *rte;
888 : TupleDesc tupDesc;
889 : List *attnums;
890 : ListCell *cur;
891 :
892 : Assert(!stmt->query);
893 :
894 : /* Open and lock the relation, using the appropriate lock type. */
895 3956 : rel = table_openrv(stmt->relation, lockmode);
896 :
897 3954 : relid = RelationGetRelid(rel);
898 :
899 3954 : nsitem = addRangeTableEntryForRelation(pstate, rel, lockmode,
900 : NULL, false, false);
901 3954 : rte = nsitem->p_rte;
902 3954 : rte->requiredPerms = (is_from ? ACL_INSERT : ACL_SELECT);
903 :
904 3954 : if (stmt->whereClause)
905 : {
906 : /* add nsitem to query namespace */
907 28 : addNSItemToQuery(pstate, nsitem, false, true, true);
908 :
909 : /* Transform the raw expression tree */
910 28 : whereClause = transformExpr(pstate, stmt->whereClause, EXPR_KIND_COPY_WHERE);
911 :
912 : /* Make sure it yields a boolean result. */
913 8 : whereClause = coerce_to_boolean(pstate, whereClause, "WHERE");
914 :
915 : /* we have to fix its collations too */
916 8 : assign_expr_collations(pstate, whereClause);
917 :
918 8 : whereClause = eval_const_expressions(NULL, whereClause);
919 :
920 8 : whereClause = (Node *) canonicalize_qual((Expr *) whereClause, false);
921 8 : whereClause = (Node *) make_ands_implicit((Expr *) whereClause);
922 : }
923 :
924 3934 : tupDesc = RelationGetDescr(rel);
925 3934 : attnums = CopyGetAttnums(tupDesc, rel, stmt->attlist);
926 18818 : foreach(cur, attnums)
927 : {
928 14924 : int attno = lfirst_int(cur) -
929 : FirstLowInvalidHeapAttributeNumber;
930 :
931 14924 : if (is_from)
932 3922 : rte->insertedCols = bms_add_member(rte->insertedCols, attno);
933 : else
934 11002 : rte->selectedCols = bms_add_member(rte->selectedCols, attno);
935 : }
936 3894 : ExecCheckRTPerms(pstate->p_rtable, true);
937 :
938 : /*
939 : * Permission check for row security policies.
940 : *
941 : * check_enable_rls will ereport(ERROR) if the user has requested
942 : * something invalid and will otherwise indicate if we should enable
943 : * RLS (returns RLS_ENABLED) or not for this COPY statement.
944 : *
945 : * If the relation has a row security policy and we are to apply it
946 : * then perform a "query" copy and allow the normal query processing
947 : * to handle the policies.
948 : *
949 : * If RLS is not enabled for this, then just fall through to the
950 : * normal non-filtering relation handling.
951 : */
952 3846 : if (check_enable_rls(rte->relid, InvalidOid, false) == RLS_ENABLED)
953 : {
954 : SelectStmt *select;
955 : ColumnRef *cr;
956 : ResTarget *target;
957 : RangeVar *from;
958 36 : List *targetList = NIL;
959 :
960 36 : if (is_from)
961 4 : ereport(ERROR,
962 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
963 : errmsg("COPY FROM not supported with row-level security"),
964 : errhint("Use INSERT statements instead.")));
965 :
966 : /*
967 : * Build target list
968 : *
969 : * If no columns are specified in the attribute list of the COPY
970 : * command, then the target list is 'all' columns. Therefore, '*'
971 : * should be used as the target list for the resulting SELECT
972 : * statement.
973 : *
974 : * In the case that columns are specified in the attribute list,
975 : * create a ColumnRef and ResTarget for each column and add them
976 : * to the target list for the resulting SELECT statement.
977 : */
978 32 : if (!stmt->attlist)
979 : {
980 8 : cr = makeNode(ColumnRef);
981 8 : cr->fields = list_make1(makeNode(A_Star));
982 8 : cr->location = -1;
983 :
984 8 : target = makeNode(ResTarget);
985 8 : target->name = NULL;
986 8 : target->indirection = NIL;
987 8 : target->val = (Node *) cr;
988 8 : target->location = -1;
989 :
990 8 : targetList = list_make1(target);
991 : }
992 : else
993 : {
994 : ListCell *lc;
995 :
996 68 : foreach(lc, stmt->attlist)
997 : {
998 : /*
999 : * Build the ColumnRef for each column. The ColumnRef
1000 : * 'fields' property is a String 'Value' node (see
1001 : * nodes/value.h) that corresponds to the column name
1002 : * respectively.
1003 : */
1004 44 : cr = makeNode(ColumnRef);
1005 44 : cr->fields = list_make1(lfirst(lc));
1006 44 : cr->location = -1;
1007 :
1008 : /* Build the ResTarget and add the ColumnRef to it. */
1009 44 : target = makeNode(ResTarget);
1010 44 : target->name = NULL;
1011 44 : target->indirection = NIL;
1012 44 : target->val = (Node *) cr;
1013 44 : target->location = -1;
1014 :
1015 : /* Add each column to the SELECT statement's target list */
1016 44 : targetList = lappend(targetList, target);
1017 : }
1018 : }
1019 :
1020 : /*
1021 : * Build RangeVar for from clause, fully qualified based on the
1022 : * relation which we have opened and locked.
1023 : */
1024 32 : from = makeRangeVar(get_namespace_name(RelationGetNamespace(rel)),
1025 32 : pstrdup(RelationGetRelationName(rel)),
1026 : -1);
1027 :
1028 : /* Build query */
1029 32 : select = makeNode(SelectStmt);
1030 32 : select->targetList = targetList;
1031 32 : select->fromClause = list_make1(from);
1032 :
1033 32 : query = makeNode(RawStmt);
1034 32 : query->stmt = (Node *) select;
1035 32 : query->stmt_location = stmt_location;
1036 32 : query->stmt_len = stmt_len;
1037 :
1038 : /*
1039 : * Close the relation for now, but keep the lock on it to prevent
1040 : * changes between now and when we start the query-based COPY.
1041 : *
1042 : * We'll reopen it later as part of the query-based COPY.
1043 : */
1044 32 : table_close(rel, NoLock);
1045 32 : rel = NULL;
1046 : }
1047 : }
1048 : else
1049 : {
1050 : Assert(stmt->query);
1051 :
1052 190 : query = makeNode(RawStmt);
1053 190 : query->stmt = stmt->query;
1054 190 : query->stmt_location = stmt_location;
1055 190 : query->stmt_len = stmt_len;
1056 :
1057 190 : relid = InvalidOid;
1058 190 : rel = NULL;
1059 : }
1060 :
1061 4024 : if (is_from)
1062 : {
1063 : Assert(rel);
1064 :
1065 : /* check read-only transaction and parallel mode */
1066 982 : if (XactReadOnly && !rel->rd_islocaltemp)
1067 0 : PreventCommandIfReadOnly("COPY FROM");
1068 :
1069 982 : cstate = BeginCopyFrom(pstate, rel, stmt->filename, stmt->is_program,
1070 : NULL, stmt->attlist, stmt->options);
1071 974 : cstate->whereClause = whereClause;
1072 974 : *processed = CopyFrom(cstate); /* copy from file to database */
1073 896 : EndCopyFrom(cstate);
1074 : }
1075 : else
1076 : {
1077 9126 : cstate = BeginCopyTo(pstate, rel, query, relid,
1078 3042 : stmt->filename, stmt->is_program,
1079 : stmt->attlist, stmt->options);
1080 2952 : *processed = DoCopyTo(cstate); /* copy from database to file */
1081 2952 : EndCopyTo(cstate);
1082 : }
1083 :
1084 3848 : if (rel != NULL)
1085 3708 : table_close(rel, NoLock);
1086 3848 : }
1087 :
1088 : /*
1089 : * Process the statement option list for COPY.
1090 : *
1091 : * Scan the options list (a list of DefElem) and transpose the information
1092 : * into cstate, applying appropriate error checking.
1093 : *
1094 : * cstate is assumed to be filled with zeroes initially.
1095 : *
1096 : * This is exported so that external users of the COPY API can sanity-check
1097 : * a list of options. In that usage, cstate should be passed as NULL
1098 : * (since external users don't know sizeof(CopyStateData)) and the collected
1099 : * data is just leaked until CurrentMemoryContext is reset.
1100 : *
1101 : * Note that additional checking, such as whether column names listed in FORCE
1102 : * QUOTE actually exist, has to be applied later. This just checks for
1103 : * self-consistency of the options list.
1104 : */
1105 : void
1106 4256 : ProcessCopyOptions(ParseState *pstate,
1107 : CopyState cstate,
1108 : bool is_from,
1109 : List *options)
1110 : {
1111 4256 : bool format_specified = false;
1112 : ListCell *option;
1113 :
1114 : /* Support external use for option sanity checking */
1115 4256 : if (cstate == NULL)
1116 82 : cstate = (CopyStateData *) palloc0(sizeof(CopyStateData));
1117 :
1118 4256 : cstate->is_copy_from = is_from;
1119 :
1120 4256 : cstate->file_encoding = -1;
1121 :
1122 : /* Extract options from the statement node tree */
1123 5050 : foreach(option, options)
1124 : {
1125 796 : DefElem *defel = lfirst_node(DefElem, option);
1126 :
1127 796 : if (strcmp(defel->defname, "format") == 0)
1128 : {
1129 274 : char *fmt = defGetString(defel);
1130 :
1131 274 : if (format_specified)
1132 0 : ereport(ERROR,
1133 : (errcode(ERRCODE_SYNTAX_ERROR),
1134 : errmsg("conflicting or redundant options"),
1135 : parser_errposition(pstate, defel->location)));
1136 274 : format_specified = true;
1137 274 : if (strcmp(fmt, "text") == 0)
1138 : /* default format */ ;
1139 232 : else if (strcmp(fmt, "csv") == 0)
1140 216 : cstate->csv_mode = true;
1141 16 : else if (strcmp(fmt, "binary") == 0)
1142 14 : cstate->binary = true;
1143 : else
1144 2 : ereport(ERROR,
1145 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1146 : errmsg("COPY format \"%s\" not recognized", fmt),
1147 : parser_errposition(pstate, defel->location)));
1148 : }
1149 522 : else if (strcmp(defel->defname, "freeze") == 0)
1150 : {
1151 32 : if (cstate->freeze)
1152 0 : ereport(ERROR,
1153 : (errcode(ERRCODE_SYNTAX_ERROR),
1154 : errmsg("conflicting or redundant options"),
1155 : parser_errposition(pstate, defel->location)));
1156 32 : cstate->freeze = defGetBoolean(defel);
1157 : }
1158 490 : else if (strcmp(defel->defname, "delimiter") == 0)
1159 : {
1160 166 : if (cstate->delim)
1161 0 : ereport(ERROR,
1162 : (errcode(ERRCODE_SYNTAX_ERROR),
1163 : errmsg("conflicting or redundant options"),
1164 : parser_errposition(pstate, defel->location)));
1165 166 : cstate->delim = defGetString(defel);
1166 : }
1167 324 : else if (strcmp(defel->defname, "null") == 0)
1168 : {
1169 92 : if (cstate->null_print)
1170 0 : ereport(ERROR,
1171 : (errcode(ERRCODE_SYNTAX_ERROR),
1172 : errmsg("conflicting or redundant options"),
1173 : parser_errposition(pstate, defel->location)));
1174 92 : cstate->null_print = defGetString(defel);
1175 : }
1176 232 : else if (strcmp(defel->defname, "header") == 0)
1177 : {
1178 42 : if (cstate->header_line)
1179 0 : ereport(ERROR,
1180 : (errcode(ERRCODE_SYNTAX_ERROR),
1181 : errmsg("conflicting or redundant options"),
1182 : parser_errposition(pstate, defel->location)));
1183 42 : cstate->header_line = defGetBoolean(defel);
1184 : }
1185 190 : else if (strcmp(defel->defname, "quote") == 0)
1186 : {
1187 56 : if (cstate->quote)
1188 0 : ereport(ERROR,
1189 : (errcode(ERRCODE_SYNTAX_ERROR),
1190 : errmsg("conflicting or redundant options"),
1191 : parser_errposition(pstate, defel->location)));
1192 56 : cstate->quote = defGetString(defel);
1193 : }
1194 134 : else if (strcmp(defel->defname, "escape") == 0)
1195 : {
1196 54 : if (cstate->escape)
1197 0 : ereport(ERROR,
1198 : (errcode(ERRCODE_SYNTAX_ERROR),
1199 : errmsg("conflicting or redundant options"),
1200 : parser_errposition(pstate, defel->location)));
1201 54 : cstate->escape = defGetString(defel);
1202 : }
1203 80 : else if (strcmp(defel->defname, "force_quote") == 0)
1204 : {
1205 28 : if (cstate->force_quote || cstate->force_quote_all)
1206 0 : ereport(ERROR,
1207 : (errcode(ERRCODE_SYNTAX_ERROR),
1208 : errmsg("conflicting or redundant options"),
1209 : parser_errposition(pstate, defel->location)));
1210 28 : if (defel->arg && IsA(defel->arg, A_Star))
1211 12 : cstate->force_quote_all = true;
1212 16 : else if (defel->arg && IsA(defel->arg, List))
1213 16 : cstate->force_quote = castNode(List, defel->arg);
1214 : else
1215 0 : ereport(ERROR,
1216 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1217 : errmsg("argument to option \"%s\" must be a list of column names",
1218 : defel->defname),
1219 : parser_errposition(pstate, defel->location)));
1220 : }
1221 52 : else if (strcmp(defel->defname, "force_not_null") == 0)
1222 : {
1223 20 : if (cstate->force_notnull)
1224 0 : ereport(ERROR,
1225 : (errcode(ERRCODE_SYNTAX_ERROR),
1226 : errmsg("conflicting or redundant options"),
1227 : parser_errposition(pstate, defel->location)));
1228 20 : if (defel->arg && IsA(defel->arg, List))
1229 20 : cstate->force_notnull = castNode(List, defel->arg);
1230 : else
1231 0 : ereport(ERROR,
1232 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1233 : errmsg("argument to option \"%s\" must be a list of column names",
1234 : defel->defname),
1235 : parser_errposition(pstate, defel->location)));
1236 : }
1237 32 : else if (strcmp(defel->defname, "force_null") == 0)
1238 : {
1239 20 : if (cstate->force_null)
1240 0 : ereport(ERROR,
1241 : (errcode(ERRCODE_SYNTAX_ERROR),
1242 : errmsg("conflicting or redundant options")));
1243 20 : if (defel->arg && IsA(defel->arg, List))
1244 20 : cstate->force_null = castNode(List, defel->arg);
1245 : else
1246 0 : ereport(ERROR,
1247 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1248 : errmsg("argument to option \"%s\" must be a list of column names",
1249 : defel->defname),
1250 : parser_errposition(pstate, defel->location)));
1251 : }
1252 12 : else if (strcmp(defel->defname, "convert_selectively") == 0)
1253 : {
1254 : /*
1255 : * Undocumented, not-accessible-from-SQL option: convert only the
1256 : * named columns to binary form, storing the rest as NULLs. It's
1257 : * allowed for the column list to be NIL.
1258 : */
1259 4 : if (cstate->convert_selectively)
1260 0 : ereport(ERROR,
1261 : (errcode(ERRCODE_SYNTAX_ERROR),
1262 : errmsg("conflicting or redundant options"),
1263 : parser_errposition(pstate, defel->location)));
1264 4 : cstate->convert_selectively = true;
1265 4 : if (defel->arg == NULL || IsA(defel->arg, List))
1266 4 : cstate->convert_select = castNode(List, defel->arg);
1267 : else
1268 0 : ereport(ERROR,
1269 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1270 : errmsg("argument to option \"%s\" must be a list of column names",
1271 : defel->defname),
1272 : parser_errposition(pstate, defel->location)));
1273 : }
1274 8 : else if (strcmp(defel->defname, "encoding") == 0)
1275 : {
1276 8 : if (cstate->file_encoding >= 0)
1277 0 : ereport(ERROR,
1278 : (errcode(ERRCODE_SYNTAX_ERROR),
1279 : errmsg("conflicting or redundant options"),
1280 : parser_errposition(pstate, defel->location)));
1281 8 : cstate->file_encoding = pg_char_to_encoding(defGetString(defel));
1282 8 : if (cstate->file_encoding < 0)
1283 0 : ereport(ERROR,
1284 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1285 : errmsg("argument to option \"%s\" must be a valid encoding name",
1286 : defel->defname),
1287 : parser_errposition(pstate, defel->location)));
1288 : }
1289 : else
1290 0 : ereport(ERROR,
1291 : (errcode(ERRCODE_SYNTAX_ERROR),
1292 : errmsg("option \"%s\" not recognized",
1293 : defel->defname),
1294 : parser_errposition(pstate, defel->location)));
1295 : }
1296 :
1297 : /*
1298 : * Check for incompatible options (must do these two before inserting
1299 : * defaults)
1300 : */
1301 4254 : if (cstate->binary && cstate->delim)
1302 0 : ereport(ERROR,
1303 : (errcode(ERRCODE_SYNTAX_ERROR),
1304 : errmsg("cannot specify DELIMITER in BINARY mode")));
1305 :
1306 4254 : if (cstate->binary && cstate->null_print)
1307 0 : ereport(ERROR,
1308 : (errcode(ERRCODE_SYNTAX_ERROR),
1309 : errmsg("cannot specify NULL in BINARY mode")));
1310 :
1311 : /* Set defaults for omitted options */
1312 4254 : if (!cstate->delim)
1313 4088 : cstate->delim = cstate->csv_mode ? "," : "\t";
1314 :
1315 4254 : if (!cstate->null_print)
1316 4162 : cstate->null_print = cstate->csv_mode ? "" : "\\N";
1317 4254 : cstate->null_print_len = strlen(cstate->null_print);
1318 :
1319 4254 : if (cstate->csv_mode)
1320 : {
1321 216 : if (!cstate->quote)
1322 164 : cstate->quote = "\"";
1323 216 : if (!cstate->escape)
1324 168 : cstate->escape = cstate->quote;
1325 : }
1326 :
1327 : /* Only single-byte delimiter strings are supported. */
1328 4254 : if (strlen(cstate->delim) != 1)
1329 2 : ereport(ERROR,
1330 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1331 : errmsg("COPY delimiter must be a single one-byte character")));
1332 :
1333 : /* Disallow end-of-line characters */
1334 4252 : if (strchr(cstate->delim, '\r') != NULL ||
1335 4252 : strchr(cstate->delim, '\n') != NULL)
1336 2 : ereport(ERROR,
1337 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1338 : errmsg("COPY delimiter cannot be newline or carriage return")));
1339 :
1340 4250 : if (strchr(cstate->null_print, '\r') != NULL ||
1341 4250 : strchr(cstate->null_print, '\n') != NULL)
1342 2 : ereport(ERROR,
1343 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1344 : errmsg("COPY null representation cannot use newline or carriage return")));
1345 :
1346 : /*
1347 : * Disallow unsafe delimiter characters in non-CSV mode. We can't allow
1348 : * backslash because it would be ambiguous. We can't allow the other
1349 : * cases because data characters matching the delimiter must be
1350 : * backslashed, and certain backslash combinations are interpreted
1351 : * non-literally by COPY IN. Disallowing all lower case ASCII letters is
1352 : * more than strictly necessary, but seems best for consistency and
1353 : * future-proofing. Likewise we disallow all digits though only octal
1354 : * digits are actually dangerous.
1355 : */
1356 4248 : if (!cstate->csv_mode &&
1357 4038 : strchr("\\.abcdefghijklmnopqrstuvwxyz0123456789",
1358 4038 : cstate->delim[0]) != NULL)
1359 10 : ereport(ERROR,
1360 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1361 : errmsg("COPY delimiter cannot be \"%s\"", cstate->delim)));
1362 :
1363 : /* Check header */
1364 4238 : if (!cstate->csv_mode && cstate->header_line)
1365 4 : ereport(ERROR,
1366 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1367 : errmsg("COPY HEADER available only in CSV mode")));
1368 :
1369 : /* Check quote */
1370 4234 : if (!cstate->csv_mode && cstate->quote != NULL)
1371 4 : ereport(ERROR,
1372 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1373 : errmsg("COPY quote available only in CSV mode")));
1374 :
1375 4230 : if (cstate->csv_mode && strlen(cstate->quote) != 1)
1376 2 : ereport(ERROR,
1377 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1378 : errmsg("COPY quote must be a single one-byte character")));
1379 :
1380 4228 : if (cstate->csv_mode && cstate->delim[0] == cstate->quote[0])
1381 2 : ereport(ERROR,
1382 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1383 : errmsg("COPY delimiter and quote must be different")));
1384 :
1385 : /* Check escape */
1386 4226 : if (!cstate->csv_mode && cstate->escape != NULL)
1387 6 : ereport(ERROR,
1388 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1389 : errmsg("COPY escape available only in CSV mode")));
1390 :
1391 4220 : if (cstate->csv_mode && strlen(cstate->escape) != 1)
1392 2 : ereport(ERROR,
1393 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1394 : errmsg("COPY escape must be a single one-byte character")));
1395 :
1396 : /* Check force_quote */
1397 4218 : if (!cstate->csv_mode && (cstate->force_quote || cstate->force_quote_all))
1398 0 : ereport(ERROR,
1399 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1400 : errmsg("COPY force quote available only in CSV mode")));
1401 4218 : if ((cstate->force_quote || cstate->force_quote_all) && is_from)
1402 0 : ereport(ERROR,
1403 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1404 : errmsg("COPY force quote only available using COPY TO")));
1405 :
1406 : /* Check force_notnull */
1407 4218 : if (!cstate->csv_mode && cstate->force_notnull != NIL)
1408 2 : ereport(ERROR,
1409 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1410 : errmsg("COPY force not null available only in CSV mode")));
1411 4216 : if (cstate->force_notnull != NIL && !is_from)
1412 0 : ereport(ERROR,
1413 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1414 : errmsg("COPY force not null only available using COPY FROM")));
1415 :
1416 : /* Check force_null */
1417 4216 : if (!cstate->csv_mode && cstate->force_null != NIL)
1418 0 : ereport(ERROR,
1419 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1420 : errmsg("COPY force null available only in CSV mode")));
1421 :
1422 4216 : if (cstate->force_null != NIL && !is_from)
1423 0 : ereport(ERROR,
1424 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1425 : errmsg("COPY force null only available using COPY FROM")));
1426 :
1427 : /* Don't allow the delimiter to appear in the null string. */
1428 4216 : if (strchr(cstate->null_print, cstate->delim[0]) != NULL)
1429 2 : ereport(ERROR,
1430 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1431 : errmsg("COPY delimiter must not appear in the NULL specification")));
1432 :
1433 : /* Don't allow the CSV quote char to appear in the null string. */
1434 4214 : if (cstate->csv_mode &&
1435 202 : strchr(cstate->null_print, cstate->quote[0]) != NULL)
1436 2 : ereport(ERROR,
1437 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1438 : errmsg("CSV quote character must not appear in the NULL specification")));
1439 4212 : }
1440 :
1441 : /*
1442 : * Common setup routines used by BeginCopyFrom and BeginCopyTo.
1443 : *
1444 : * Iff <binary>, unload or reload in the binary format, as opposed to the
1445 : * more wasteful but more robust and portable text format.
1446 : *
1447 : * Iff <oids>, unload or reload the format that includes OID information.
1448 : * On input, we accept OIDs whether or not the table has an OID column,
1449 : * but silently drop them if it does not. On output, we report an error
1450 : * if the user asks for OIDs in a table that has none (not providing an
1451 : * OID column might seem friendlier, but could seriously confuse programs).
1452 : *
1453 : * If in the text format, delimit columns with delimiter <delim> and print
1454 : * NULL values as <null_print>.
1455 : */
1456 : static CopyState
1457 4174 : BeginCopy(ParseState *pstate,
1458 : bool is_from,
1459 : Relation rel,
1460 : RawStmt *raw_query,
1461 : Oid queryRelId,
1462 : List *attnamelist,
1463 : List *options)
1464 : {
1465 : CopyState cstate;
1466 : TupleDesc tupDesc;
1467 : int num_phys_attrs;
1468 : MemoryContext oldcontext;
1469 :
1470 : /* Allocate workspace and zero all fields */
1471 4174 : cstate = (CopyStateData *) palloc0(sizeof(CopyStateData));
1472 :
1473 : /*
1474 : * We allocate everything used by a cstate in a new memory context. This
1475 : * avoids memory leaks during repeated use of COPY in a query.
1476 : */
1477 4174 : cstate->copycontext = AllocSetContextCreate(CurrentMemoryContext,
1478 : "COPY",
1479 : ALLOCSET_DEFAULT_SIZES);
1480 :
1481 4174 : oldcontext = MemoryContextSwitchTo(cstate->copycontext);
1482 :
1483 : /* Extract options from the statement node tree */
1484 4174 : ProcessCopyOptions(pstate, cstate, is_from, options);
1485 :
1486 : /* Process the source/target relation or query */
1487 4172 : if (rel)
1488 : {
1489 : Assert(!raw_query);
1490 :
1491 3950 : cstate->rel = rel;
1492 :
1493 3950 : tupDesc = RelationGetDescr(cstate->rel);
1494 : }
1495 : else
1496 : {
1497 : List *rewritten;
1498 : Query *query;
1499 : PlannedStmt *plan;
1500 : DestReceiver *dest;
1501 :
1502 : Assert(!is_from);
1503 222 : cstate->rel = NULL;
1504 :
1505 : /*
1506 : * Run parse analysis and rewrite. Note this also acquires sufficient
1507 : * locks on the source table(s).
1508 : *
1509 : * Because the parser and planner tend to scribble on their input, we
1510 : * make a preliminary copy of the source querytree. This prevents
1511 : * problems in the case that the COPY is in a portal or plpgsql
1512 : * function and is executed repeatedly. (See also the same hack in
1513 : * DECLARE CURSOR and PREPARE.) XXX FIXME someday.
1514 : */
1515 222 : rewritten = pg_analyze_and_rewrite(copyObject(raw_query),
1516 : pstate->p_sourcetext, NULL, 0,
1517 : NULL);
1518 :
1519 : /* check that we got back something we can work with */
1520 214 : if (rewritten == NIL)
1521 : {
1522 12 : ereport(ERROR,
1523 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1524 : errmsg("DO INSTEAD NOTHING rules are not supported for COPY")));
1525 : }
1526 202 : else if (list_length(rewritten) > 1)
1527 : {
1528 : ListCell *lc;
1529 :
1530 : /* examine queries to determine which error message to issue */
1531 68 : foreach(lc, rewritten)
1532 : {
1533 56 : Query *q = lfirst_node(Query, lc);
1534 :
1535 56 : if (q->querySource == QSRC_QUAL_INSTEAD_RULE)
1536 12 : ereport(ERROR,
1537 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1538 : errmsg("conditional DO INSTEAD rules are not supported for COPY")));
1539 44 : if (q->querySource == QSRC_NON_INSTEAD_RULE)
1540 12 : ereport(ERROR,
1541 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1542 : errmsg("DO ALSO rules are not supported for the COPY")));
1543 : }
1544 :
1545 12 : ereport(ERROR,
1546 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1547 : errmsg("multi-statement DO INSTEAD rules are not supported for COPY")));
1548 : }
1549 :
1550 166 : query = linitial_node(Query, rewritten);
1551 :
1552 : /* The grammar allows SELECT INTO, but we don't support that */
1553 166 : if (query->utilityStmt != NULL &&
1554 8 : IsA(query->utilityStmt, CreateTableAsStmt))
1555 8 : ereport(ERROR,
1556 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1557 : errmsg("COPY (SELECT INTO) is not supported")));
1558 :
1559 : Assert(query->utilityStmt == NULL);
1560 :
1561 : /*
1562 : * Similarly the grammar doesn't enforce the presence of a RETURNING
1563 : * clause, but this is required here.
1564 : */
1565 158 : if (query->commandType != CMD_SELECT &&
1566 48 : query->returningList == NIL)
1567 : {
1568 : Assert(query->commandType == CMD_INSERT ||
1569 : query->commandType == CMD_UPDATE ||
1570 : query->commandType == CMD_DELETE);
1571 :
1572 12 : ereport(ERROR,
1573 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1574 : errmsg("COPY query must have a RETURNING clause")));
1575 : }
1576 :
1577 : /* plan the query */
1578 146 : plan = pg_plan_query(query, pstate->p_sourcetext,
1579 : CURSOR_OPT_PARALLEL_OK, NULL);
1580 :
1581 : /*
1582 : * With row level security and a user using "COPY relation TO", we
1583 : * have to convert the "COPY relation TO" to a query-based COPY (eg:
1584 : * "COPY (SELECT * FROM relation) TO"), to allow the rewriter to add
1585 : * in any RLS clauses.
1586 : *
1587 : * When this happens, we are passed in the relid of the originally
1588 : * found relation (which we have locked). As the planner will look up
1589 : * the relation again, we double-check here to make sure it found the
1590 : * same one that we have locked.
1591 : */
1592 144 : if (queryRelId != InvalidOid)
1593 : {
1594 : /*
1595 : * Note that with RLS involved there may be multiple relations,
1596 : * and while the one we need is almost certainly first, we don't
1597 : * make any guarantees of that in the planner, so check the whole
1598 : * list and make sure we find the original relation.
1599 : */
1600 32 : if (!list_member_oid(plan->relationOids, queryRelId))
1601 0 : ereport(ERROR,
1602 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1603 : errmsg("relation referenced by COPY statement has changed")));
1604 : }
1605 :
1606 : /*
1607 : * Use a snapshot with an updated command ID to ensure this query sees
1608 : * results of any previously executed queries.
1609 : */
1610 144 : PushCopiedSnapshot(GetActiveSnapshot());
1611 144 : UpdateActiveSnapshotCommandId();
1612 :
1613 : /* Create dest receiver for COPY OUT */
1614 144 : dest = CreateDestReceiver(DestCopyOut);
1615 144 : ((DR_copy *) dest)->cstate = cstate;
1616 :
1617 : /* Create a QueryDesc requesting no output */
1618 144 : cstate->queryDesc = CreateQueryDesc(plan, pstate->p_sourcetext,
1619 : GetActiveSnapshot(),
1620 : InvalidSnapshot,
1621 : dest, NULL, NULL, 0);
1622 :
1623 : /*
1624 : * Call ExecutorStart to prepare the plan for execution.
1625 : *
1626 : * ExecutorStart computes a result tupdesc for us
1627 : */
1628 144 : ExecutorStart(cstate->queryDesc, 0);
1629 :
1630 140 : tupDesc = cstate->queryDesc->tupDesc;
1631 : }
1632 :
1633 : /* Generate or convert list of attributes to process */
1634 4090 : cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
1635 :
1636 4090 : num_phys_attrs = tupDesc->natts;
1637 :
1638 : /* Convert FORCE_QUOTE name list to per-column flags, check validity */
1639 4090 : cstate->force_quote_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
1640 4090 : if (cstate->force_quote_all)
1641 : {
1642 : int i;
1643 :
1644 36 : for (i = 0; i < num_phys_attrs; i++)
1645 24 : cstate->force_quote_flags[i] = true;
1646 : }
1647 4078 : else if (cstate->force_quote)
1648 : {
1649 : List *attnums;
1650 : ListCell *cur;
1651 :
1652 16 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->force_quote);
1653 :
1654 32 : foreach(cur, attnums)
1655 : {
1656 16 : int attnum = lfirst_int(cur);
1657 16 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1658 :
1659 16 : if (!list_member_int(cstate->attnumlist, attnum))
1660 0 : ereport(ERROR,
1661 : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1662 : errmsg("FORCE_QUOTE column \"%s\" not referenced by COPY",
1663 : NameStr(attr->attname))));
1664 16 : cstate->force_quote_flags[attnum - 1] = true;
1665 : }
1666 : }
1667 :
1668 : /* Convert FORCE_NOT_NULL name list to per-column flags, check validity */
1669 4090 : cstate->force_notnull_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
1670 4090 : if (cstate->force_notnull)
1671 : {
1672 : List *attnums;
1673 : ListCell *cur;
1674 :
1675 18 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->force_notnull);
1676 :
1677 36 : foreach(cur, attnums)
1678 : {
1679 22 : int attnum = lfirst_int(cur);
1680 22 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1681 :
1682 22 : if (!list_member_int(cstate->attnumlist, attnum))
1683 4 : ereport(ERROR,
1684 : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1685 : errmsg("FORCE_NOT_NULL column \"%s\" not referenced by COPY",
1686 : NameStr(attr->attname))));
1687 18 : cstate->force_notnull_flags[attnum - 1] = true;
1688 : }
1689 : }
1690 :
1691 : /* Convert FORCE_NULL name list to per-column flags, check validity */
1692 4086 : cstate->force_null_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
1693 4086 : if (cstate->force_null)
1694 : {
1695 : List *attnums;
1696 : ListCell *cur;
1697 :
1698 18 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->force_null);
1699 :
1700 36 : foreach(cur, attnums)
1701 : {
1702 22 : int attnum = lfirst_int(cur);
1703 22 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1704 :
1705 22 : if (!list_member_int(cstate->attnumlist, attnum))
1706 4 : ereport(ERROR,
1707 : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1708 : errmsg("FORCE_NULL column \"%s\" not referenced by COPY",
1709 : NameStr(attr->attname))));
1710 18 : cstate->force_null_flags[attnum - 1] = true;
1711 : }
1712 : }
1713 :
1714 : /* Convert convert_selectively name list to per-column flags */
1715 4082 : if (cstate->convert_selectively)
1716 : {
1717 : List *attnums;
1718 : ListCell *cur;
1719 :
1720 4 : cstate->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
1721 :
1722 4 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->convert_select);
1723 :
1724 8 : foreach(cur, attnums)
1725 : {
1726 4 : int attnum = lfirst_int(cur);
1727 4 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1728 :
1729 4 : if (!list_member_int(cstate->attnumlist, attnum))
1730 0 : ereport(ERROR,
1731 : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1732 : errmsg_internal("selected column \"%s\" not referenced by COPY",
1733 : NameStr(attr->attname))));
1734 4 : cstate->convert_select_flags[attnum - 1] = true;
1735 : }
1736 : }
1737 :
1738 : /* Use client encoding when ENCODING option is not specified. */
1739 4082 : if (cstate->file_encoding < 0)
1740 4074 : cstate->file_encoding = pg_get_client_encoding();
1741 :
1742 : /*
1743 : * Set up encoding conversion info. Even if the file and server encodings
1744 : * are the same, we must apply pg_any_to_server() to validate data in
1745 : * multibyte encodings.
1746 : */
1747 4082 : cstate->need_transcoding =
1748 8154 : (cstate->file_encoding != GetDatabaseEncoding() ||
1749 4072 : pg_database_encoding_max_length() > 1);
1750 : /* See Multibyte encoding comment above */
1751 4082 : cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding);
1752 :
1753 4082 : cstate->copy_dest = COPY_FILE; /* default */
1754 :
1755 4082 : MemoryContextSwitchTo(oldcontext);
1756 :
1757 4082 : return cstate;
1758 : }
1759 :
1760 : /*
1761 : * Closes the pipe to an external program, checking the pclose() return code.
1762 : */
1763 : static void
1764 0 : ClosePipeToProgram(CopyState cstate)
1765 : {
1766 : int pclose_rc;
1767 :
1768 : Assert(cstate->is_program);
1769 :
1770 0 : pclose_rc = ClosePipeStream(cstate->copy_file);
1771 0 : if (pclose_rc == -1)
1772 0 : ereport(ERROR,
1773 : (errcode_for_file_access(),
1774 : errmsg("could not close pipe to external command: %m")));
1775 0 : else if (pclose_rc != 0)
1776 : {
1777 : /*
1778 : * If we ended a COPY FROM PROGRAM before reaching EOF, then it's
1779 : * expectable for the called program to fail with SIGPIPE, and we
1780 : * should not report that as an error. Otherwise, SIGPIPE indicates a
1781 : * problem.
1782 : */
1783 0 : if (cstate->is_copy_from && !cstate->reached_eof &&
1784 0 : wait_result_is_signal(pclose_rc, SIGPIPE))
1785 0 : return;
1786 :
1787 0 : ereport(ERROR,
1788 : (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
1789 : errmsg("program \"%s\" failed",
1790 : cstate->filename),
1791 : errdetail_internal("%s", wait_result_to_str(pclose_rc))));
1792 : }
1793 : }
1794 :
1795 : /*
1796 : * Release resources allocated in a cstate for COPY TO/FROM.
1797 : */
1798 : static void
1799 3900 : EndCopy(CopyState cstate)
1800 : {
1801 3900 : if (cstate->is_program)
1802 : {
1803 0 : ClosePipeToProgram(cstate);
1804 : }
1805 : else
1806 : {
1807 3900 : if (cstate->filename != NULL && FreeFile(cstate->copy_file))
1808 0 : ereport(ERROR,
1809 : (errcode_for_file_access(),
1810 : errmsg("could not close file \"%s\": %m",
1811 : cstate->filename)));
1812 : }
1813 :
1814 3900 : MemoryContextDelete(cstate->copycontext);
1815 3900 : pfree(cstate);
1816 3900 : }
1817 :
1818 : /*
1819 : * Setup CopyState to read tuples from a table or a query for COPY TO.
1820 : */
1821 : static CopyState
1822 3042 : BeginCopyTo(ParseState *pstate,
1823 : Relation rel,
1824 : RawStmt *query,
1825 : Oid queryRelId,
1826 : const char *filename,
1827 : bool is_program,
1828 : List *attnamelist,
1829 : List *options)
1830 : {
1831 : CopyState cstate;
1832 3042 : bool pipe = (filename == NULL);
1833 : MemoryContext oldcontext;
1834 :
1835 3042 : if (rel != NULL && rel->rd_rel->relkind != RELKIND_RELATION)
1836 : {
1837 8 : if (rel->rd_rel->relkind == RELKIND_VIEW)
1838 8 : ereport(ERROR,
1839 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1840 : errmsg("cannot copy from view \"%s\"",
1841 : RelationGetRelationName(rel)),
1842 : errhint("Try the COPY (SELECT ...) TO variant.")));
1843 0 : else if (rel->rd_rel->relkind == RELKIND_MATVIEW)
1844 0 : ereport(ERROR,
1845 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1846 : errmsg("cannot copy from materialized view \"%s\"",
1847 : RelationGetRelationName(rel)),
1848 : errhint("Try the COPY (SELECT ...) TO variant.")));
1849 0 : else if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
1850 0 : ereport(ERROR,
1851 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1852 : errmsg("cannot copy from foreign table \"%s\"",
1853 : RelationGetRelationName(rel)),
1854 : errhint("Try the COPY (SELECT ...) TO variant.")));
1855 0 : else if (rel->rd_rel->relkind == RELKIND_SEQUENCE)
1856 0 : ereport(ERROR,
1857 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1858 : errmsg("cannot copy from sequence \"%s\"",
1859 : RelationGetRelationName(rel))));
1860 0 : else if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
1861 0 : ereport(ERROR,
1862 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1863 : errmsg("cannot copy from partitioned table \"%s\"",
1864 : RelationGetRelationName(rel)),
1865 : errhint("Try the COPY (SELECT ...) TO variant.")));
1866 : else
1867 0 : ereport(ERROR,
1868 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1869 : errmsg("cannot copy from non-table relation \"%s\"",
1870 : RelationGetRelationName(rel))));
1871 : }
1872 :
1873 3034 : cstate = BeginCopy(pstate, false, rel, query, queryRelId, attnamelist,
1874 : options);
1875 2952 : oldcontext = MemoryContextSwitchTo(cstate->copycontext);
1876 :
1877 2952 : if (pipe)
1878 : {
1879 : Assert(!is_program); /* the grammar does not allow this */
1880 2926 : if (whereToSendOutput != DestRemote)
1881 0 : cstate->copy_file = stdout;
1882 : }
1883 : else
1884 : {
1885 26 : cstate->filename = pstrdup(filename);
1886 26 : cstate->is_program = is_program;
1887 :
1888 26 : if (is_program)
1889 : {
1890 0 : cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_W);
1891 0 : if (cstate->copy_file == NULL)
1892 0 : ereport(ERROR,
1893 : (errcode_for_file_access(),
1894 : errmsg("could not execute command \"%s\": %m",
1895 : cstate->filename)));
1896 : }
1897 : else
1898 : {
1899 : mode_t oumask; /* Pre-existing umask value */
1900 : struct stat st;
1901 :
1902 : /*
1903 : * Prevent write to relative path ... too easy to shoot oneself in
1904 : * the foot by overwriting a database file ...
1905 : */
1906 26 : if (!is_absolute_path(filename))
1907 0 : ereport(ERROR,
1908 : (errcode(ERRCODE_INVALID_NAME),
1909 : errmsg("relative path not allowed for COPY to file")));
1910 :
1911 26 : oumask = umask(S_IWGRP | S_IWOTH);
1912 26 : PG_TRY();
1913 : {
1914 26 : cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_W);
1915 : }
1916 0 : PG_FINALLY();
1917 : {
1918 26 : umask(oumask);
1919 : }
1920 26 : PG_END_TRY();
1921 26 : if (cstate->copy_file == NULL)
1922 : {
1923 : /* copy errno because ereport subfunctions might change it */
1924 0 : int save_errno = errno;
1925 :
1926 0 : ereport(ERROR,
1927 : (errcode_for_file_access(),
1928 : errmsg("could not open file \"%s\" for writing: %m",
1929 : cstate->filename),
1930 : (save_errno == ENOENT || save_errno == EACCES) ?
1931 : errhint("COPY TO instructs the PostgreSQL server process to write a file. "
1932 : "You may want a client-side facility such as psql's \\copy.") : 0));
1933 : }
1934 :
1935 26 : if (fstat(fileno(cstate->copy_file), &st))
1936 0 : ereport(ERROR,
1937 : (errcode_for_file_access(),
1938 : errmsg("could not stat file \"%s\": %m",
1939 : cstate->filename)));
1940 :
1941 26 : if (S_ISDIR(st.st_mode))
1942 0 : ereport(ERROR,
1943 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1944 : errmsg("\"%s\" is a directory", cstate->filename)));
1945 : }
1946 : }
1947 :
1948 2952 : MemoryContextSwitchTo(oldcontext);
1949 :
1950 2952 : return cstate;
1951 : }
1952 :
1953 : /*
1954 : * This intermediate routine exists mainly to localize the effects of setjmp
1955 : * so we don't need to plaster a lot of variables with "volatile".
1956 : */
1957 : static uint64
1958 2952 : DoCopyTo(CopyState cstate)
1959 : {
1960 2952 : bool pipe = (cstate->filename == NULL);
1961 2952 : bool fe_copy = (pipe && whereToSendOutput == DestRemote);
1962 : uint64 processed;
1963 :
1964 2952 : PG_TRY();
1965 : {
1966 2952 : if (fe_copy)
1967 2926 : SendCopyBegin(cstate);
1968 :
1969 2952 : processed = CopyTo(cstate);
1970 :
1971 2952 : if (fe_copy)
1972 2926 : SendCopyEnd(cstate);
1973 : }
1974 0 : PG_CATCH();
1975 : {
1976 : /*
1977 : * Make sure we turn off old-style COPY OUT mode upon error. It is
1978 : * okay to do this in all cases, since it does nothing if the mode is
1979 : * not on.
1980 : */
1981 0 : pq_endcopyout(true);
1982 0 : PG_RE_THROW();
1983 : }
1984 2952 : PG_END_TRY();
1985 :
1986 2952 : return processed;
1987 : }
1988 :
1989 : /*
1990 : * Clean up storage and release resources for COPY TO.
1991 : */
1992 : static void
1993 2952 : EndCopyTo(CopyState cstate)
1994 : {
1995 2952 : if (cstate->queryDesc != NULL)
1996 : {
1997 : /* Close down the query and free resources. */
1998 140 : ExecutorFinish(cstate->queryDesc);
1999 140 : ExecutorEnd(cstate->queryDesc);
2000 140 : FreeQueryDesc(cstate->queryDesc);
2001 140 : PopActiveSnapshot();
2002 : }
2003 :
2004 : /* Clean up storage */
2005 2952 : EndCopy(cstate);
2006 2952 : }
2007 :
2008 : /*
2009 : * Copy from relation or query TO file.
2010 : */
2011 : static uint64
2012 2952 : CopyTo(CopyState cstate)
2013 : {
2014 : TupleDesc tupDesc;
2015 : int num_phys_attrs;
2016 : ListCell *cur;
2017 : uint64 processed;
2018 :
2019 2952 : if (cstate->rel)
2020 2812 : tupDesc = RelationGetDescr(cstate->rel);
2021 : else
2022 140 : tupDesc = cstate->queryDesc->tupDesc;
2023 2952 : num_phys_attrs = tupDesc->natts;
2024 2952 : cstate->null_print_client = cstate->null_print; /* default */
2025 :
2026 : /* We use fe_msgbuf as a per-row buffer regardless of copy_dest */
2027 2952 : cstate->fe_msgbuf = makeStringInfo();
2028 :
2029 : /* Get info about the columns we need to process. */
2030 2952 : cstate->out_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
2031 14054 : foreach(cur, cstate->attnumlist)
2032 : {
2033 11102 : int attnum = lfirst_int(cur);
2034 : Oid out_func_oid;
2035 : bool isvarlena;
2036 11102 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
2037 :
2038 11102 : if (cstate->binary)
2039 28 : getTypeBinaryOutputInfo(attr->atttypid,
2040 : &out_func_oid,
2041 : &isvarlena);
2042 : else
2043 11074 : getTypeOutputInfo(attr->atttypid,
2044 : &out_func_oid,
2045 : &isvarlena);
2046 11102 : fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
2047 : }
2048 :
2049 : /*
2050 : * Create a temporary memory context that we can reset once per row to
2051 : * recover palloc'd memory. This avoids any problems with leaks inside
2052 : * datatype output routines, and should be faster than retail pfree's
2053 : * anyway. (We don't need a whole econtext as CopyFrom does.)
2054 : */
2055 2952 : cstate->rowcontext = AllocSetContextCreate(CurrentMemoryContext,
2056 : "COPY TO",
2057 : ALLOCSET_DEFAULT_SIZES);
2058 :
2059 2952 : if (cstate->binary)
2060 : {
2061 : /* Generate header for a binary copy */
2062 : int32 tmp;
2063 :
2064 : /* Signature */
2065 4 : CopySendData(cstate, BinarySignature, 11);
2066 : /* Flags field */
2067 4 : tmp = 0;
2068 4 : CopySendInt32(cstate, tmp);
2069 : /* No header extension */
2070 4 : tmp = 0;
2071 4 : CopySendInt32(cstate, tmp);
2072 : }
2073 : else
2074 : {
2075 : /*
2076 : * For non-binary copy, we need to convert null_print to file
2077 : * encoding, because it will be sent directly with CopySendString.
2078 : */
2079 2948 : if (cstate->need_transcoding)
2080 2938 : cstate->null_print_client = pg_server_to_any(cstate->null_print,
2081 : cstate->null_print_len,
2082 : cstate->file_encoding);
2083 :
2084 : /* if a header has been requested send the line */
2085 2948 : if (cstate->header_line)
2086 : {
2087 8 : bool hdr_delim = false;
2088 :
2089 24 : foreach(cur, cstate->attnumlist)
2090 : {
2091 16 : int attnum = lfirst_int(cur);
2092 : char *colname;
2093 :
2094 16 : if (hdr_delim)
2095 8 : CopySendChar(cstate, cstate->delim[0]);
2096 16 : hdr_delim = true;
2097 :
2098 16 : colname = NameStr(TupleDescAttr(tupDesc, attnum - 1)->attname);
2099 :
2100 16 : CopyAttributeOutCSV(cstate, colname, false,
2101 16 : list_length(cstate->attnumlist) == 1);
2102 : }
2103 :
2104 8 : CopySendEndOfRow(cstate);
2105 : }
2106 : }
2107 :
2108 2952 : if (cstate->rel)
2109 : {
2110 : TupleTableSlot *slot;
2111 : TableScanDesc scandesc;
2112 :
2113 2812 : scandesc = table_beginscan(cstate->rel, GetActiveSnapshot(), 0, NULL);
2114 2812 : slot = table_slot_create(cstate->rel, NULL);
2115 :
2116 2812 : processed = 0;
2117 1471580 : while (table_scan_getnextslot(scandesc, ForwardScanDirection, slot))
2118 : {
2119 1468768 : CHECK_FOR_INTERRUPTS();
2120 :
2121 : /* Deconstruct the tuple ... */
2122 1468768 : slot_getallattrs(slot);
2123 :
2124 : /* Format and send the data */
2125 1468768 : CopyOneRowTo(cstate, slot);
2126 1468768 : processed++;
2127 : }
2128 :
2129 2812 : ExecDropSingleTupleTableSlot(slot);
2130 2812 : table_endscan(scandesc);
2131 : }
2132 : else
2133 : {
2134 : /* run the plan --- the dest receiver will send tuples */
2135 140 : ExecutorRun(cstate->queryDesc, ForwardScanDirection, 0L, true);
2136 140 : processed = ((DR_copy *) cstate->queryDesc->dest)->processed;
2137 : }
2138 :
2139 2952 : if (cstate->binary)
2140 : {
2141 : /* Generate trailer for a binary copy */
2142 4 : CopySendInt16(cstate, -1);
2143 : /* Need to flush out the trailer */
2144 4 : CopySendEndOfRow(cstate);
2145 : }
2146 :
2147 2952 : MemoryContextDelete(cstate->rowcontext);
2148 :
2149 2952 : return processed;
2150 : }
2151 :
2152 : /*
2153 : * Emit one row during CopyTo().
2154 : */
2155 : static void
2156 1473236 : CopyOneRowTo(CopyState cstate, TupleTableSlot *slot)
2157 : {
2158 1473236 : bool need_delim = false;
2159 1473236 : FmgrInfo *out_functions = cstate->out_functions;
2160 : MemoryContext oldcontext;
2161 : ListCell *cur;
2162 : char *string;
2163 :
2164 1473236 : MemoryContextReset(cstate->rowcontext);
2165 1473236 : oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
2166 :
2167 1473236 : if (cstate->binary)
2168 : {
2169 : /* Binary per-tuple header */
2170 12 : CopySendInt16(cstate, list_length(cstate->attnumlist));
2171 : }
2172 :
2173 : /* Make sure the tuple is fully deconstructed */
2174 1473236 : slot_getallattrs(slot);
2175 :
2176 6316310 : foreach(cur, cstate->attnumlist)
2177 : {
2178 4843074 : int attnum = lfirst_int(cur);
2179 4843074 : Datum value = slot->tts_values[attnum - 1];
2180 4843074 : bool isnull = slot->tts_isnull[attnum - 1];
2181 :
2182 4843074 : if (!cstate->binary)
2183 : {
2184 4842990 : if (need_delim)
2185 3369846 : CopySendChar(cstate, cstate->delim[0]);
2186 4842990 : need_delim = true;
2187 : }
2188 :
2189 4843074 : if (isnull)
2190 : {
2191 282074 : if (!cstate->binary)
2192 282054 : CopySendString(cstate, cstate->null_print_client);
2193 : else
2194 20 : CopySendInt32(cstate, -1);
2195 : }
2196 : else
2197 : {
2198 4561000 : if (!cstate->binary)
2199 : {
2200 4560936 : string = OutputFunctionCall(&out_functions[attnum - 1],
2201 : value);
2202 4560936 : if (cstate->csv_mode)
2203 760 : CopyAttributeOutCSV(cstate, string,
2204 380 : cstate->force_quote_flags[attnum - 1],
2205 380 : list_length(cstate->attnumlist) == 1);
2206 : else
2207 4560556 : CopyAttributeOutText(cstate, string);
2208 : }
2209 : else
2210 : {
2211 : bytea *outputbytes;
2212 :
2213 64 : outputbytes = SendFunctionCall(&out_functions[attnum - 1],
2214 : value);
2215 64 : CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
2216 64 : CopySendData(cstate, VARDATA(outputbytes),
2217 64 : VARSIZE(outputbytes) - VARHDRSZ);
2218 : }
2219 : }
2220 : }
2221 :
2222 1473236 : CopySendEndOfRow(cstate);
2223 :
2224 1473236 : MemoryContextSwitchTo(oldcontext);
2225 1473236 : }
2226 :
2227 :
2228 : /*
2229 : * error context callback for COPY FROM
2230 : *
2231 : * The argument for the error context must be CopyState.
2232 : */
2233 : void
2234 106 : CopyFromErrorCallback(void *arg)
2235 : {
2236 106 : CopyState cstate = (CopyState) arg;
2237 : char curlineno_str[32];
2238 :
2239 106 : snprintf(curlineno_str, sizeof(curlineno_str), UINT64_FORMAT,
2240 : cstate->cur_lineno);
2241 :
2242 106 : if (cstate->binary)
2243 : {
2244 : /* can't usefully display the data */
2245 0 : if (cstate->cur_attname)
2246 0 : errcontext("COPY %s, line %s, column %s",
2247 : cstate->cur_relname, curlineno_str,
2248 : cstate->cur_attname);
2249 : else
2250 0 : errcontext("COPY %s, line %s",
2251 : cstate->cur_relname, curlineno_str);
2252 : }
2253 : else
2254 : {
2255 106 : if (cstate->cur_attname && cstate->cur_attval)
2256 14 : {
2257 : /* error is relevant to a particular column */
2258 : char *attval;
2259 :
2260 14 : attval = limit_printout_length(cstate->cur_attval);
2261 14 : errcontext("COPY %s, line %s, column %s: \"%s\"",
2262 : cstate->cur_relname, curlineno_str,
2263 : cstate->cur_attname, attval);
2264 14 : pfree(attval);
2265 : }
2266 92 : else if (cstate->cur_attname)
2267 : {
2268 : /* error is relevant to a particular column, value is NULL */
2269 4 : errcontext("COPY %s, line %s, column %s: null input",
2270 : cstate->cur_relname, curlineno_str,
2271 : cstate->cur_attname);
2272 : }
2273 : else
2274 : {
2275 : /*
2276 : * Error is relevant to a particular line.
2277 : *
2278 : * If line_buf still contains the correct line, and it's already
2279 : * transcoded, print it. If it's still in a foreign encoding, it's
2280 : * quite likely that the error is precisely a failure to do
2281 : * encoding conversion (ie, bad data). We dare not try to convert
2282 : * it, and at present there's no way to regurgitate it without
2283 : * conversion. So we have to punt and just report the line number.
2284 : */
2285 88 : if (cstate->line_buf_valid &&
2286 84 : (cstate->line_buf_converted || !cstate->need_transcoding))
2287 84 : {
2288 : char *lineval;
2289 :
2290 84 : lineval = limit_printout_length(cstate->line_buf.data);
2291 84 : errcontext("COPY %s, line %s: \"%s\"",
2292 : cstate->cur_relname, curlineno_str, lineval);
2293 84 : pfree(lineval);
2294 : }
2295 : else
2296 : {
2297 4 : errcontext("COPY %s, line %s",
2298 : cstate->cur_relname, curlineno_str);
2299 : }
2300 : }
2301 : }
2302 106 : }
2303 :
2304 : /*
2305 : * Make sure we don't print an unreasonable amount of COPY data in a message.
2306 : *
2307 : * It would seem a lot easier to just use the sprintf "precision" limit to
2308 : * truncate the string. However, some versions of glibc have a bug/misfeature
2309 : * that vsnprintf will always fail (return -1) if it is asked to truncate
2310 : * a string that contains invalid byte sequences for the current encoding.
2311 : * So, do our own truncation. We return a pstrdup'd copy of the input.
2312 : */
2313 : static char *
2314 98 : limit_printout_length(const char *str)
2315 : {
2316 : #define MAX_COPY_DATA_DISPLAY 100
2317 :
2318 98 : int slen = strlen(str);
2319 : int len;
2320 : char *res;
2321 :
2322 : /* Fast path if definitely okay */
2323 98 : if (slen <= MAX_COPY_DATA_DISPLAY)
2324 98 : return pstrdup(str);
2325 :
2326 : /* Apply encoding-dependent truncation */
2327 0 : len = pg_mbcliplen(str, slen, MAX_COPY_DATA_DISPLAY);
2328 :
2329 : /*
2330 : * Truncate, and add "..." to show we truncated the input.
2331 : */
2332 0 : res = (char *) palloc(len + 4);
2333 0 : memcpy(res, str, len);
2334 0 : strcpy(res + len, "...");
2335 :
2336 0 : return res;
2337 : }
2338 :
2339 : /*
2340 : * Allocate memory and initialize a new CopyMultiInsertBuffer for this
2341 : * ResultRelInfo.
2342 : */
2343 : static CopyMultiInsertBuffer *
2344 954 : CopyMultiInsertBufferInit(ResultRelInfo *rri)
2345 : {
2346 : CopyMultiInsertBuffer *buffer;
2347 :
2348 954 : buffer = (CopyMultiInsertBuffer *) palloc(sizeof(CopyMultiInsertBuffer));
2349 954 : memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES);
2350 954 : buffer->resultRelInfo = rri;
2351 954 : buffer->bistate = GetBulkInsertState();
2352 954 : buffer->nused = 0;
2353 :
2354 954 : return buffer;
2355 : }
2356 :
2357 : /*
2358 : * Make a new buffer for this ResultRelInfo.
2359 : */
2360 : static inline void
2361 954 : CopyMultiInsertInfoSetupBuffer(CopyMultiInsertInfo *miinfo,
2362 : ResultRelInfo *rri)
2363 : {
2364 : CopyMultiInsertBuffer *buffer;
2365 :
2366 954 : buffer = CopyMultiInsertBufferInit(rri);
2367 :
2368 : /* Setup back-link so we can easily find this buffer again */
2369 954 : rri->ri_CopyMultiInsertBuffer = buffer;
2370 : /* Record that we're tracking this buffer */
2371 954 : miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
2372 954 : }
2373 :
2374 : /*
2375 : * Initialize an already allocated CopyMultiInsertInfo.
2376 : *
2377 : * If rri is a non-partitioned table then a CopyMultiInsertBuffer is set up
2378 : * for that table.
2379 : */
2380 : static void
2381 956 : CopyMultiInsertInfoInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
2382 : CopyState cstate, EState *estate, CommandId mycid,
2383 : int ti_options)
2384 : {
2385 956 : miinfo->multiInsertBuffers = NIL;
2386 956 : miinfo->bufferedTuples = 0;
2387 956 : miinfo->bufferedBytes = 0;
2388 956 : miinfo->cstate = cstate;
2389 956 : miinfo->estate = estate;
2390 956 : miinfo->mycid = mycid;
2391 956 : miinfo->ti_options = ti_options;
2392 :
2393 : /*
2394 : * Only setup the buffer when not dealing with a partitioned table.
2395 : * Buffers for partitioned tables will just be setup when we need to send
2396 : * tuples their way for the first time.
2397 : */
2398 956 : if (rri->ri_RelationDesc->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
2399 922 : CopyMultiInsertInfoSetupBuffer(miinfo, rri);
2400 956 : }
2401 :
2402 : /*
2403 : * Returns true if the buffers are full
2404 : */
2405 : static inline bool
2406 1155548 : CopyMultiInsertInfoIsFull(CopyMultiInsertInfo *miinfo)
2407 : {
2408 1155548 : if (miinfo->bufferedTuples >= MAX_BUFFERED_TUPLES ||
2409 1154748 : miinfo->bufferedBytes >= MAX_BUFFERED_BYTES)
2410 870 : return true;
2411 1154678 : return false;
2412 : }
2413 :
2414 : /*
2415 : * Returns true if we have no buffered tuples
2416 : */
2417 : static inline bool
2418 938 : CopyMultiInsertInfoIsEmpty(CopyMultiInsertInfo *miinfo)
2419 : {
2420 938 : return miinfo->bufferedTuples == 0;
2421 : }
2422 :
2423 : /*
2424 : * Write the tuples stored in 'buffer' out to the table.
2425 : */
2426 : static inline void
2427 1836 : CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo,
2428 : CopyMultiInsertBuffer *buffer)
2429 : {
2430 : MemoryContext oldcontext;
2431 : int i;
2432 : uint64 save_cur_lineno;
2433 1836 : CopyState cstate = miinfo->cstate;
2434 1836 : EState *estate = miinfo->estate;
2435 1836 : CommandId mycid = miinfo->mycid;
2436 1836 : int ti_options = miinfo->ti_options;
2437 1836 : bool line_buf_valid = cstate->line_buf_valid;
2438 1836 : int nused = buffer->nused;
2439 1836 : ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
2440 1836 : TupleTableSlot **slots = buffer->slots;
2441 :
2442 : /* Set es_result_relation_info to the ResultRelInfo we're flushing. */
2443 1836 : estate->es_result_relation_info = resultRelInfo;
2444 :
2445 : /*
2446 : * Print error context information correctly, if one of the operations
2447 : * below fail.
2448 : */
2449 1836 : cstate->line_buf_valid = false;
2450 1836 : save_cur_lineno = cstate->cur_lineno;
2451 :
2452 : /*
2453 : * table_multi_insert may leak memory, so switch to short-lived memory
2454 : * context before calling it.
2455 : */
2456 1836 : oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
2457 1836 : table_multi_insert(resultRelInfo->ri_RelationDesc,
2458 : slots,
2459 : nused,
2460 : mycid,
2461 : ti_options,
2462 : buffer->bistate);
2463 1836 : MemoryContextSwitchTo(oldcontext);
2464 :
2465 1157290 : for (i = 0; i < nused; i++)
2466 : {
2467 : /*
2468 : * If there are any indexes, update them for all the inserted tuples,
2469 : * and run AFTER ROW INSERT triggers.
2470 : */
2471 1155458 : if (resultRelInfo->ri_NumIndices > 0)
2472 : {
2473 : List *recheckIndexes;
2474 :
2475 208598 : cstate->cur_lineno = buffer->linenos[i];
2476 : recheckIndexes =
2477 208598 : ExecInsertIndexTuples(buffer->slots[i], estate, false, NULL,
2478 : NIL);
2479 417188 : ExecARInsertTriggers(estate, resultRelInfo,
2480 208594 : slots[i], recheckIndexes,
2481 : cstate->transition_capture);
2482 208594 : list_free(recheckIndexes);
2483 : }
2484 :
2485 : /*
2486 : * There's no indexes, but see if we need to run AFTER ROW INSERT
2487 : * triggers anyway.
2488 : */
2489 946860 : else if (resultRelInfo->ri_TrigDesc != NULL &&
2490 32 : (resultRelInfo->ri_TrigDesc->trig_insert_after_row ||
2491 20 : resultRelInfo->ri_TrigDesc->trig_insert_new_table))
2492 : {
2493 24 : cstate->cur_lineno = buffer->linenos[i];
2494 48 : ExecARInsertTriggers(estate, resultRelInfo,
2495 24 : slots[i], NIL, cstate->transition_capture);
2496 : }
2497 :
2498 1155454 : ExecClearTuple(slots[i]);
2499 : }
2500 :
2501 : /* Mark that all slots are free */
2502 1832 : buffer->nused = 0;
2503 :
2504 : /* reset cur_lineno and line_buf_valid to what they were */
2505 1832 : cstate->line_buf_valid = line_buf_valid;
2506 1832 : cstate->cur_lineno = save_cur_lineno;
2507 1832 : }
2508 :
2509 : /*
2510 : * Drop used slots and free member for this buffer.
2511 : *
2512 : * The buffer must be flushed before cleanup.
2513 : */
2514 : static inline void
2515 912 : CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
2516 : CopyMultiInsertBuffer *buffer)
2517 : {
2518 : int i;
2519 :
2520 : /* Ensure buffer was flushed */
2521 : Assert(buffer->nused == 0);
2522 :
2523 : /* Remove back-link to ourself */
2524 912 : buffer->resultRelInfo->ri_CopyMultiInsertBuffer = NULL;
2525 :
2526 912 : FreeBulkInsertState(buffer->bistate);
2527 :
2528 : /* Since we only create slots on demand, just drop the non-null ones. */
2529 390626 : for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++)
2530 389714 : ExecDropSingleTupleTableSlot(buffer->slots[i]);
2531 :
2532 912 : table_finish_bulk_insert(buffer->resultRelInfo->ri_RelationDesc,
2533 : miinfo->ti_options);
2534 :
2535 912 : pfree(buffer);
2536 912 : }
2537 :
2538 : /*
2539 : * Write out all stored tuples in all buffers out to the tables.
2540 : *
2541 : * Once flushed we also trim the tracked buffers list down to size by removing
2542 : * the buffers created earliest first.
2543 : *
2544 : * Callers should pass 'curr_rri' is the ResultRelInfo that's currently being
2545 : * used. When cleaning up old buffers we'll never remove the one for
2546 : * 'curr_rri'.
2547 : */
2548 : static inline void
2549 1628 : CopyMultiInsertInfoFlush(CopyMultiInsertInfo *miinfo, ResultRelInfo *curr_rri)
2550 : {
2551 : ListCell *lc;
2552 :
2553 3460 : foreach(lc, miinfo->multiInsertBuffers)
2554 : {
2555 1836 : CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) lfirst(lc);
2556 :
2557 1836 : CopyMultiInsertBufferFlush(miinfo, buffer);
2558 : }
2559 :
2560 1624 : miinfo->bufferedTuples = 0;
2561 1624 : miinfo->bufferedBytes = 0;
2562 :
2563 : /*
2564 : * Trim the list of tracked buffers down if it exceeds the limit. Here we
2565 : * remove buffers starting with the ones we created first. It seems more
2566 : * likely that these older ones are less likely to be needed than ones
2567 : * that were just created.
2568 : */
2569 1624 : while (list_length(miinfo->multiInsertBuffers) > MAX_PARTITION_BUFFERS)
2570 : {
2571 : CopyMultiInsertBuffer *buffer;
2572 :
2573 0 : buffer = (CopyMultiInsertBuffer *) linitial(miinfo->multiInsertBuffers);
2574 :
2575 : /*
2576 : * We never want to remove the buffer that's currently being used, so
2577 : * if we happen to find that then move it to the end of the list.
2578 : */
2579 0 : if (buffer->resultRelInfo == curr_rri)
2580 : {
2581 0 : miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
2582 0 : miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
2583 0 : buffer = (CopyMultiInsertBuffer *) linitial(miinfo->multiInsertBuffers);
2584 : }
2585 :
2586 0 : CopyMultiInsertBufferCleanup(miinfo, buffer);
2587 0 : miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
2588 : }
2589 1624 : }
2590 :
2591 : /*
2592 : * Cleanup allocated buffers and free memory
2593 : */
2594 : static inline void
2595 914 : CopyMultiInsertInfoCleanup(CopyMultiInsertInfo *miinfo)
2596 : {
2597 : ListCell *lc;
2598 :
2599 1826 : foreach(lc, miinfo->multiInsertBuffers)
2600 912 : CopyMultiInsertBufferCleanup(miinfo, lfirst(lc));
2601 :
2602 914 : list_free(miinfo->multiInsertBuffers);
2603 914 : }
2604 :
2605 : /*
2606 : * Get the next TupleTableSlot that the next tuple should be stored in.
2607 : *
2608 : * Callers must ensure that the buffer is not full.
2609 : */
2610 : static inline TupleTableSlot *
2611 1156470 : CopyMultiInsertInfoNextFreeSlot(CopyMultiInsertInfo *miinfo,
2612 : ResultRelInfo *rri)
2613 : {
2614 1156470 : CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer;
2615 1156470 : int nused = buffer->nused;
2616 :
2617 : Assert(buffer != NULL);
2618 : Assert(nused < MAX_BUFFERED_TUPLES);
2619 :
2620 1156470 : if (buffer->slots[nused] == NULL)
2621 389848 : buffer->slots[nused] = table_slot_create(rri->ri_RelationDesc, NULL);
2622 1156470 : return buffer->slots[nused];
2623 : }
2624 :
2625 : /*
2626 : * Record the previously reserved TupleTableSlot that was reserved by
2627 : * CopyMultiInsertInfoNextFreeSlot as being consumed.
2628 : */
2629 : static inline void
2630 1155548 : CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
2631 : TupleTableSlot *slot, int tuplen, uint64 lineno)
2632 : {
2633 1155548 : CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer;
2634 :
2635 : Assert(buffer != NULL);
2636 : Assert(slot == buffer->slots[buffer->nused]);
2637 :
2638 : /* Store the line number so we can properly report any errors later */
2639 1155548 : buffer->linenos[buffer->nused] = lineno;
2640 :
2641 : /* Record this slot as being used */
2642 1155548 : buffer->nused++;
2643 :
2644 : /* Update how many tuples are stored and their size */
2645 1155548 : miinfo->bufferedTuples++;
2646 1155548 : miinfo->bufferedBytes += tuplen;
2647 1155548 : }
2648 :
2649 : /*
2650 : * Copy FROM file to relation.
2651 : */
2652 : uint64
2653 1076 : CopyFrom(CopyState cstate)
2654 : {
2655 : ResultRelInfo *resultRelInfo;
2656 : ResultRelInfo *target_resultRelInfo;
2657 1076 : ResultRelInfo *prevResultRelInfo = NULL;
2658 1076 : EState *estate = CreateExecutorState(); /* for ExecConstraints() */
2659 : ModifyTableState *mtstate;
2660 : ExprContext *econtext;
2661 1076 : TupleTableSlot *singleslot = NULL;
2662 1076 : MemoryContext oldcontext = CurrentMemoryContext;
2663 :
2664 1076 : PartitionTupleRouting *proute = NULL;
2665 : ErrorContextCallback errcallback;
2666 1076 : CommandId mycid = GetCurrentCommandId(true);
2667 1076 : int ti_options = 0; /* start with default options for insert */
2668 1076 : BulkInsertState bistate = NULL;
2669 : CopyInsertMethod insertMethod;
2670 1076 : CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
2671 1076 : uint64 processed = 0;
2672 : bool has_before_insert_row_trig;
2673 : bool has_instead_insert_row_trig;
2674 1076 : bool leafpart_use_multi_insert = false;
2675 :
2676 : Assert(cstate->rel);
2677 :
2678 : /*
2679 : * The target must be a plain, foreign, or partitioned relation, or have
2680 : * an INSTEAD OF INSERT row trigger. (Currently, such triggers are only
2681 : * allowed on views, so we only hint about them in the view case.)
2682 : */
2683 1076 : if (cstate->rel->rd_rel->relkind != RELKIND_RELATION &&
2684 86 : cstate->rel->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
2685 62 : cstate->rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE &&
2686 12 : !(cstate->rel->trigdesc &&
2687 8 : cstate->rel->trigdesc->trig_insert_instead_row))
2688 : {
2689 4 : if (cstate->rel->rd_rel->relkind == RELKIND_VIEW)
2690 4 : ereport(ERROR,
2691 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
2692 : errmsg("cannot copy to view \"%s\"",
2693 : RelationGetRelationName(cstate->rel)),
2694 : errhint("To enable copying to a view, provide an INSTEAD OF INSERT trigger.")));
2695 0 : else if (cstate->rel->rd_rel->relkind == RELKIND_MATVIEW)
2696 0 : ereport(ERROR,
2697 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
2698 : errmsg("cannot copy to materialized view \"%s\"",
2699 : RelationGetRelationName(cstate->rel))));
2700 0 : else if (cstate->rel->rd_rel->relkind == RELKIND_SEQUENCE)
2701 0 : ereport(ERROR,
2702 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
2703 : errmsg("cannot copy to sequence \"%s\"",
2704 : RelationGetRelationName(cstate->rel))));
2705 : else
2706 0 : ereport(ERROR,
2707 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
2708 : errmsg("cannot copy to non-table relation \"%s\"",
2709 : RelationGetRelationName(cstate->rel))));
2710 : }
2711 :
2712 : /*
2713 : * If the target file is new-in-transaction, we assume that checking FSM
2714 : * for free space is a waste of time. This could possibly be wrong, but
2715 : * it's unlikely.
2716 : */
2717 1072 : if (RELKIND_HAS_STORAGE(cstate->rel->rd_rel->relkind) &&
2718 990 : (cstate->rel->rd_createSubid != InvalidSubTransactionId ||
2719 986 : cstate->rel->rd_firstRelfilenodeSubid != InvalidSubTransactionId))
2720 38 : ti_options |= TABLE_INSERT_SKIP_FSM;
2721 :
2722 : /*
2723 : * Optimize if new relfilenode was created in this subxact or one of its
2724 : * committed children and we won't see those rows later as part of an
2725 : * earlier scan or command. The subxact test ensures that if this subxact
2726 : * aborts then the frozen rows won't be visible after xact cleanup. Note
2727 : * that the stronger test of exactly which subtransaction created it is
2728 : * crucial for correctness of this optimization. The test for an earlier
2729 : * scan or command tolerates false negatives. FREEZE causes other sessions
2730 : * to see rows they would not see under MVCC, and a false negative merely
2731 : * spreads that anomaly to the current session.
2732 : */
2733 1072 : if (cstate->freeze)
2734 : {
2735 : /*
2736 : * We currently disallow COPY FREEZE on partitioned tables. The
2737 : * reason for this is that we've simply not yet opened the partitions
2738 : * to determine if the optimization can be applied to them. We could
2739 : * go and open them all here, but doing so may be quite a costly
2740 : * overhead for small copies. In any case, we may just end up routing
2741 : * tuples to a small number of partitions. It seems better just to
2742 : * raise an ERROR for partitioned tables.
2743 : */
2744 32 : if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
2745 : {
2746 4 : ereport(ERROR,
2747 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2748 : errmsg("cannot perform COPY FREEZE on a partitioned table")));
2749 : }
2750 :
2751 : /*
2752 : * Tolerate one registration for the benefit of FirstXactSnapshot.
2753 : * Scan-bearing queries generally create at least two registrations,
2754 : * though relying on that is fragile, as is ignoring ActiveSnapshot.
2755 : * Clear CatalogSnapshot to avoid counting its registration. We'll
2756 : * still detect ongoing catalog scans, each of which separately
2757 : * registers the snapshot it uses.
2758 : */
2759 28 : InvalidateCatalogSnapshot();
2760 28 : if (!ThereAreNoPriorRegisteredSnapshots() || !ThereAreNoReadyPortals())
2761 0 : ereport(ERROR,
2762 : (errcode(ERRCODE_INVALID_TRANSACTION_STATE),
2763 : errmsg("cannot perform COPY FREEZE because of prior transaction activity")));
2764 :
2765 56 : if (cstate->rel->rd_createSubid != GetCurrentSubTransactionId() &&
2766 28 : cstate->rel->rd_newRelfilenodeSubid != GetCurrentSubTransactionId())
2767 12 : ereport(ERROR,
2768 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2769 : errmsg("cannot perform COPY FREEZE because the table was not created or truncated in the current subtransaction")));
2770 :
2771 16 : ti_options |= TABLE_INSERT_FROZEN;
2772 : }
2773 :
2774 : /*
2775 : * We need a ResultRelInfo so we can use the regular executor's
2776 : * index-entry-making machinery. (There used to be a huge amount of code
2777 : * here that basically duplicated execUtils.c ...)
2778 : */
2779 1056 : resultRelInfo = makeNode(ResultRelInfo);
2780 1056 : InitResultRelInfo(resultRelInfo,
2781 : cstate->rel,
2782 : 1, /* must match rel's position in range_table */
2783 : NULL,
2784 : 0);
2785 1056 : target_resultRelInfo = resultRelInfo;
2786 :
2787 : /* Verify the named relation is a valid target for INSERT */
2788 1056 : CheckValidResultRel(resultRelInfo, CMD_INSERT);
2789 :
2790 1054 : ExecOpenIndices(resultRelInfo, false);
2791 :
2792 1054 : estate->es_result_relations = resultRelInfo;
2793 1054 : estate->es_num_result_relations = 1;
2794 1054 : estate->es_result_relation_info = resultRelInfo;
2795 :
2796 1054 : ExecInitRangeTable(estate, cstate->range_table);
2797 :
2798 : /*
2799 : * Set up a ModifyTableState so we can let FDW(s) init themselves for
2800 : * foreign-table result relation(s).
2801 : */
2802 1054 : mtstate = makeNode(ModifyTableState);
2803 1054 : mtstate->ps.plan = NULL;
2804 1054 : mtstate->ps.state = estate;
2805 1054 : mtstate->operation = CMD_INSERT;
2806 1054 : mtstate->resultRelInfo = estate->es_result_relations;
2807 :
2808 1054 : if (resultRelInfo->ri_FdwRoutine != NULL &&
2809 22 : resultRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL)
2810 22 : resultRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate,
2811 : resultRelInfo);
2812 :
2813 : /* Prepare to catch AFTER triggers. */
2814 1054 : AfterTriggerBeginQuery();
2815 :
2816 : /*
2817 : * If there are any triggers with transition tables on the named relation,
2818 : * we need to be prepared to capture transition tuples.
2819 : *
2820 : * Because partition tuple routing would like to know about whether
2821 : * transition capture is active, we also set it in mtstate, which is
2822 : * passed to ExecFindPartition() below.
2823 : */
2824 1054 : cstate->transition_capture = mtstate->mt_transition_capture =
2825 1054 : MakeTransitionCaptureState(cstate->rel->trigdesc,
2826 1054 : RelationGetRelid(cstate->rel),
2827 : CMD_INSERT);
2828 :
2829 : /*
2830 : * If the named relation is a partitioned table, initialize state for
2831 : * CopyFrom tuple routing.
2832 : */
2833 1054 : if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
2834 46 : proute = ExecSetupPartitionTupleRouting(estate, NULL, cstate->rel);
2835 :
2836 1054 : if (cstate->whereClause)
2837 8 : cstate->qualexpr = ExecInitQual(castNode(List, cstate->whereClause),
2838 : &mtstate->ps);
2839 :
2840 : /*
2841 : * It's generally more efficient to prepare a bunch of tuples for
2842 : * insertion, and insert them in one table_multi_insert() call, than call
2843 : * table_tuple_insert() separately for every tuple. However, there are a
2844 : * number of reasons why we might not be able to do this. These are
2845 : * explained below.
2846 : */
2847 1054 : if (resultRelInfo->ri_TrigDesc != NULL &&
2848 108 : (resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
2849 44 : resultRelInfo->ri_TrigDesc->trig_insert_instead_row))
2850 : {
2851 : /*
2852 : * Can't support multi-inserts when there are any BEFORE/INSTEAD OF
2853 : * triggers on the table. Such triggers might query the table we're
2854 : * inserting into and act differently if the tuples that have already
2855 : * been processed and prepared for insertion are not there.
2856 : */
2857 72 : insertMethod = CIM_SINGLE;
2858 : }
2859 982 : else if (proute != NULL && resultRelInfo->ri_TrigDesc != NULL &&
2860 16 : resultRelInfo->ri_TrigDesc->trig_insert_new_table)
2861 : {
2862 : /*
2863 : * For partitioned tables we can't support multi-inserts when there
2864 : * are any statement level insert triggers. It might be possible to
2865 : * allow partitioned tables with such triggers in the future, but for
2866 : * now, CopyMultiInsertInfoFlush expects that any before row insert
2867 : * and statement level insert triggers are on the same relation.
2868 : */
2869 12 : insertMethod = CIM_SINGLE;
2870 : }
2871 970 : else if (resultRelInfo->ri_FdwRoutine != NULL ||
2872 956 : cstate->volatile_defexprs)
2873 : {
2874 : /*
2875 : * Can't support multi-inserts to foreign tables or if there are any
2876 : * volatile default expressions in the table. Similarly to the
2877 : * trigger case above, such expressions may query the table we're
2878 : * inserting into.
2879 : *
2880 : * Note: It does not matter if any partitions have any volatile
2881 : * default expressions as we use the defaults from the target of the
2882 : * COPY command.
2883 : */
2884 14 : insertMethod = CIM_SINGLE;
2885 : }
2886 956 : else if (contain_volatile_functions(cstate->whereClause))
2887 : {
2888 : /*
2889 : * Can't support multi-inserts if there are any volatile function
2890 : * expressions in WHERE clause. Similarly to the trigger case above,
2891 : * such expressions may query the table we're inserting into.
2892 : */
2893 0 : insertMethod = CIM_SINGLE;
2894 : }
2895 : else
2896 : {
2897 : /*
2898 : * For partitioned tables, we may still be able to perform bulk
2899 : * inserts. However, the possibility of this depends on which types
2900 : * of triggers exist on the partition. We must disable bulk inserts
2901 : * if the partition is a foreign table or it has any before row insert
2902 : * or insert instead triggers (same as we checked above for the parent
2903 : * table). Since the partition's resultRelInfos are initialized only
2904 : * when we actually need to insert the first tuple into them, we must
2905 : * have the intermediate insert method of CIM_MULTI_CONDITIONAL to
2906 : * flag that we must later determine if we can use bulk-inserts for
2907 : * the partition being inserted into.
2908 : */
2909 956 : if (proute)
2910 34 : insertMethod = CIM_MULTI_CONDITIONAL;
2911 : else
2912 922 : insertMethod = CIM_MULTI;
2913 :
2914 956 : CopyMultiInsertInfoInit(&multiInsertInfo, resultRelInfo, cstate,
2915 : estate, mycid, ti_options);
2916 : }
2917 :
2918 : /*
2919 : * If not using batch mode (which allocates slots as needed) set up a
2920 : * tuple slot too. When inserting into a partitioned table, we also need
2921 : * one, even if we might batch insert, to read the tuple in the root
2922 : * partition's form.
2923 : */
2924 1054 : if (insertMethod == CIM_SINGLE || insertMethod == CIM_MULTI_CONDITIONAL)
2925 : {
2926 132 : singleslot = table_slot_create(resultRelInfo->ri_RelationDesc,
2927 : &estate->es_tupleTable);
2928 132 : bistate = GetBulkInsertState();
2929 : }
2930 :
2931 1162 : has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
2932 108 : resultRelInfo->ri_TrigDesc->trig_insert_before_row);
2933 :
2934 1162 : has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
2935 108 : resultRelInfo->ri_TrigDesc->trig_insert_instead_row);
2936 :
2937 : /*
2938 : * Check BEFORE STATEMENT insertion triggers. It's debatable whether we
2939 : * should do this for COPY, since it's not really an "INSERT" statement as
2940 : * such. However, executing these triggers maintains consistency with the
2941 : * EACH ROW triggers that we already fire on COPY.
2942 : */
2943 1054 : ExecBSInsertTriggers(estate, resultRelInfo);
2944 :
2945 1054 : econtext = GetPerTupleExprContext(estate);
2946 :
2947 : /* Set up callback to identify error line number */
2948 1054 : errcallback.callback = CopyFromErrorCallback;
2949 1054 : errcallback.arg = (void *) cstate;
2950 1054 : errcallback.previous = error_context_stack;
2951 1054 : error_context_stack = &errcallback;
2952 :
2953 : for (;;)
2954 1155822 : {
2955 : TupleTableSlot *myslot;
2956 : bool skip_tuple;
2957 :
2958 1156876 : CHECK_FOR_INTERRUPTS();
2959 :
2960 : /*
2961 : * Reset the per-tuple exprcontext. We do this after every tuple, to
2962 : * clean-up after expression evaluations etc.
2963 : */
2964 1156876 : ResetPerTupleExprContext(estate);
2965 :
2966 : /* select slot to (initially) load row into */
2967 1156876 : if (insertMethod == CIM_SINGLE || proute)
2968 : {
2969 208544 : myslot = singleslot;
2970 208544 : Assert(myslot != NULL);
2971 : }
2972 : else
2973 : {
2974 : Assert(resultRelInfo == target_resultRelInfo);
2975 : Assert(insertMethod == CIM_MULTI);
2976 :
2977 948332 : myslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
2978 : resultRelInfo);
2979 : }
2980 :
2981 : /*
2982 : * Switch to per-tuple context before calling NextCopyFrom, which does
2983 : * evaluate default expressions etc. and requires per-tuple context.
2984 : */
2985 1156876 : MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
2986 :
2987 1156876 : ExecClearTuple(myslot);
2988 :
2989 : /* Directly store the values/nulls array in the slot */
2990 1156876 : if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
2991 998 : break;
2992 :
2993 1155846 : ExecStoreVirtualTuple(myslot);
2994 :
2995 : /*
2996 : * Constraints and where clause might reference the tableoid column,
2997 : * so (re-)initialize tts_tableOid before evaluating them.
2998 : */
2999 1155846 : myslot->tts_tableOid = RelationGetRelid(target_resultRelInfo->ri_RelationDesc);
3000 :
3001 : /* Triggers and stuff need to be invoked in query context. */
3002 1155846 : MemoryContextSwitchTo(oldcontext);
3003 :
3004 1155846 : if (cstate->whereClause)
3005 : {
3006 32 : econtext->ecxt_scantuple = myslot;
3007 : /* Skip items that don't match COPY's WHERE clause */
3008 32 : if (!ExecQual(cstate->qualexpr, econtext))
3009 20 : continue;
3010 : }
3011 :
3012 : /* Determine the partition to insert the tuple into */
3013 1155826 : if (proute)
3014 : {
3015 : TupleConversionMap *map;
3016 :
3017 : /*
3018 : * Attempt to find a partition suitable for this tuple.
3019 : * ExecFindPartition() will raise an error if none can be found or
3020 : * if the found partition is not suitable for INSERTs.
3021 : */
3022 208232 : resultRelInfo = ExecFindPartition(mtstate, target_resultRelInfo,
3023 : proute, myslot, estate);
3024 :
3025 208230 : if (prevResultRelInfo != resultRelInfo)
3026 : {
3027 : /* Determine which triggers exist on this partition */
3028 100268 : has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
3029 36 : resultRelInfo->ri_TrigDesc->trig_insert_before_row);
3030 :
3031 100268 : has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
3032 36 : resultRelInfo->ri_TrigDesc->trig_insert_instead_row);
3033 :
3034 : /*
3035 : * Disable multi-inserts when the partition has BEFORE/INSTEAD
3036 : * OF triggers, or if the partition is a foreign partition.
3037 : */
3038 200428 : leafpart_use_multi_insert = insertMethod == CIM_MULTI_CONDITIONAL &&
3039 100196 : !has_before_insert_row_trig &&
3040 300608 : !has_instead_insert_row_trig &&
3041 100180 : resultRelInfo->ri_FdwRoutine == NULL;
3042 :
3043 : /* Set the multi-insert buffer to use for this partition. */
3044 100232 : if (leafpart_use_multi_insert)
3045 : {
3046 100176 : if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL)
3047 32 : CopyMultiInsertInfoSetupBuffer(&multiInsertInfo,
3048 : resultRelInfo);
3049 : }
3050 56 : else if (insertMethod == CIM_MULTI_CONDITIONAL &&
3051 20 : !CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
3052 : {
3053 : /*
3054 : * Flush pending inserts if this partition can't use
3055 : * batching, so rows are visible to triggers etc.
3056 : */
3057 0 : CopyMultiInsertInfoFlush(&multiInsertInfo, resultRelInfo);
3058 : }
3059 :
3060 100232 : if (bistate != NULL)
3061 100232 : ReleaseBulkInsertStatePin(bistate);
3062 100232 : prevResultRelInfo = resultRelInfo;
3063 : }
3064 :
3065 : /*
3066 : * For ExecInsertIndexTuples() to work on the partition's indexes
3067 : */
3068 208230 : estate->es_result_relation_info = resultRelInfo;
3069 :
3070 : /*
3071 : * If we're capturing transition tuples, we might need to convert
3072 : * from the partition rowtype to root rowtype.
3073 : */
3074 208230 : if (cstate->transition_capture != NULL)
3075 : {
3076 36 : if (has_before_insert_row_trig)
3077 : {
3078 : /*
3079 : * If there are any BEFORE triggers on the partition,
3080 : * we'll have to be ready to convert their result back to
3081 : * tuplestore format.
3082 : */
3083 4 : cstate->transition_capture->tcs_original_insert_tuple = NULL;
3084 8 : cstate->transition_capture->tcs_map =
3085 4 : resultRelInfo->ri_PartitionInfo->pi_PartitionToRootMap;
3086 : }
3087 : else
3088 : {
3089 : /*
3090 : * Otherwise, just remember the original unconverted
3091 : * tuple, to avoid a needless round trip conversion.
3092 : */
3093 32 : cstate->transition_capture->tcs_original_insert_tuple = myslot;
3094 32 : cstate->transition_capture->tcs_map = NULL;
3095 : }
3096 : }
3097 :
3098 : /*
3099 : * We might need to convert from the root rowtype to the partition
3100 : * rowtype.
3101 : */
3102 208230 : map = resultRelInfo->ri_PartitionInfo->pi_RootToPartitionMap;
3103 208230 : if (insertMethod == CIM_SINGLE || !leafpart_use_multi_insert)
3104 : {
3105 : /* non batch insert */
3106 166 : if (map != NULL)
3107 : {
3108 : TupleTableSlot *new_slot;
3109 :
3110 74 : new_slot = resultRelInfo->ri_PartitionInfo->pi_PartitionTupleSlot;
3111 74 : myslot = execute_attr_map_slot(map->attrMap, myslot, new_slot);
3112 : }
3113 : }
3114 : else
3115 : {
3116 : /*
3117 : * Prepare to queue up tuple for later batch insert into
3118 : * current partition.
3119 : */
3120 : TupleTableSlot *batchslot;
3121 :
3122 : /* no other path available for partitioned table */
3123 : Assert(insertMethod == CIM_MULTI_CONDITIONAL);
3124 :
3125 208138 : batchslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
3126 : resultRelInfo);
3127 :
3128 208138 : if (map != NULL)
3129 8128 : myslot = execute_attr_map_slot(map->attrMap, myslot,
3130 : batchslot);
3131 : else
3132 : {
3133 : /*
3134 : * This looks more expensive than it is (Believe me, I
3135 : * optimized it away. Twice.). The input is in virtual
3136 : * form, and we'll materialize the slot below - for most
3137 : * slot types the copy performs the work materialization
3138 : * would later require anyway.
3139 : */
3140 200010 : ExecCopySlot(batchslot, myslot);
3141 200010 : myslot = batchslot;
3142 : }
3143 : }
3144 :
3145 : /* ensure that triggers etc see the right relation */
3146 208230 : myslot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
3147 : }
3148 :
3149 1155824 : skip_tuple = false;
3150 :
3151 : /* BEFORE ROW INSERT Triggers */
3152 1155824 : if (has_before_insert_row_trig)
3153 : {
3154 188 : if (!ExecBRInsertTriggers(estate, resultRelInfo, myslot))
3155 12 : skip_tuple = true; /* "do nothing" */
3156 : }
3157 :
3158 1155824 : if (!skip_tuple)
3159 : {
3160 : /*
3161 : * If there is an INSTEAD OF INSERT ROW trigger, let it handle the
3162 : * tuple. Otherwise, proceed with inserting the tuple into the
3163 : * table or foreign table.
3164 : */
3165 1155812 : if (has_instead_insert_row_trig)
3166 : {
3167 8 : ExecIRInsertTriggers(estate, resultRelInfo, myslot);
3168 : }
3169 : else
3170 : {
3171 : /* Compute stored generated columns */
3172 1155804 : if (resultRelInfo->ri_RelationDesc->rd_att->constr &&
3173 400858 : resultRelInfo->ri_RelationDesc->rd_att->constr->has_generated_stored)
3174 22 : ExecComputeStoredGenerated(estate, myslot, CMD_INSERT);
3175 :
3176 : /*
3177 : * If the target is a plain table, check the constraints of
3178 : * the tuple.
3179 : */
3180 1155804 : if (resultRelInfo->ri_FdwRoutine == NULL &&
3181 1155764 : resultRelInfo->ri_RelationDesc->rd_att->constr)
3182 400846 : ExecConstraints(resultRelInfo, myslot, estate);
3183 :
3184 : /*
3185 : * Also check the tuple against the partition constraint, if
3186 : * there is one; except that if we got here via tuple-routing,
3187 : * we don't need to if there's no BR trigger defined on the
3188 : * partition.
3189 : */
3190 1155784 : if (resultRelInfo->ri_PartitionCheck &&
3191 208222 : (proute == NULL || has_before_insert_row_trig))
3192 54 : ExecPartitionCheck(resultRelInfo, myslot, estate, true);
3193 :
3194 : /* Store the slot in the multi-insert buffer, when enabled. */
3195 1155784 : if (insertMethod == CIM_MULTI || leafpart_use_multi_insert)
3196 : {
3197 : /*
3198 : * The slot previously might point into the per-tuple
3199 : * context. For batching it needs to be longer lived.
3200 : */
3201 1155548 : ExecMaterializeSlot(myslot);
3202 :
3203 : /* Add this tuple to the tuple buffer */
3204 1155548 : CopyMultiInsertInfoStore(&multiInsertInfo,
3205 : resultRelInfo, myslot,
3206 : cstate->line_buf.len,
3207 : cstate->cur_lineno);
3208 :
3209 : /*
3210 : * If enough inserts have queued up, then flush all
3211 : * buffers out to their tables.
3212 : */
3213 1156418 : if (CopyMultiInsertInfoIsFull(&multiInsertInfo))
3214 870 : CopyMultiInsertInfoFlush(&multiInsertInfo, resultRelInfo);
3215 : }
3216 : else
3217 : {
3218 236 : List *recheckIndexes = NIL;
3219 :
3220 : /* OK, store the tuple */
3221 236 : if (resultRelInfo->ri_FdwRoutine != NULL)
3222 : {
3223 40 : myslot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
3224 : resultRelInfo,
3225 : myslot,
3226 : NULL);
3227 :
3228 38 : if (myslot == NULL) /* "do nothing" */
3229 4 : continue; /* next tuple please */
3230 :
3231 : /*
3232 : * AFTER ROW Triggers might reference the tableoid
3233 : * column, so (re-)initialize tts_tableOid before
3234 : * evaluating them.
3235 : */
3236 34 : myslot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
3237 : }
3238 : else
3239 : {
3240 : /* OK, store the tuple and create index entries for it */
3241 196 : table_tuple_insert(resultRelInfo->ri_RelationDesc,
3242 : myslot, mycid, ti_options, bistate);
3243 :
3244 196 : if (resultRelInfo->ri_NumIndices > 0)
3245 0 : recheckIndexes = ExecInsertIndexTuples(myslot,
3246 : estate,
3247 : false,
3248 : NULL,
3249 : NIL);
3250 : }
3251 :
3252 : /* AFTER ROW INSERT Triggers */
3253 230 : ExecARInsertTriggers(estate, resultRelInfo, myslot,
3254 : recheckIndexes, cstate->transition_capture);
3255 :
3256 230 : list_free(recheckIndexes);
3257 : }
3258 : }
3259 :
3260 : /*
3261 : * We count only tuples not suppressed by a BEFORE INSERT trigger
3262 : * or FDW; this is the same definition used by nodeModifyTable.c
3263 : * for counting tuples inserted by an INSERT command.
3264 : */
3265 1155786 : processed++;
3266 : }
3267 : }
3268 :
3269 : /* Flush any remaining buffered tuples */
3270 998 : if (insertMethod != CIM_SINGLE)
3271 : {
3272 918 : if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
3273 758 : CopyMultiInsertInfoFlush(&multiInsertInfo, NULL);
3274 : }
3275 :
3276 : /* Done, clean up */
3277 994 : error_context_stack = errcallback.previous;
3278 :
3279 994 : if (bistate != NULL)
3280 112 : FreeBulkInsertState(bistate);
3281 :
3282 994 : MemoryContextSwitchTo(oldcontext);
3283 :
3284 : /*
3285 : * In the old protocol, tell pqcomm that we can process normal protocol
3286 : * messages again.
3287 : */
3288 994 : if (cstate->copy_dest == COPY_OLD_FE)
3289 0 : pq_endmsgread();
3290 :
3291 : /* Execute AFTER STATEMENT insertion triggers */
3292 994 : ExecASInsertTriggers(estate, target_resultRelInfo, cstate->transition_capture);
3293 :
3294 : /* Handle queued AFTER triggers */
3295 994 : AfterTriggerEndQuery(estate);
3296 :
3297 994 : ExecResetTupleTable(estate->es_tupleTable, false);
3298 :
3299 : /* Allow the FDW to shut down */
3300 994 : if (target_resultRelInfo->ri_FdwRoutine != NULL &&
3301 20 : target_resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL)
3302 20 : target_resultRelInfo->ri_FdwRoutine->EndForeignInsert(estate,
3303 : target_resultRelInfo);
3304 :
3305 : /* Tear down the multi-insert buffer data */
3306 994 : if (insertMethod != CIM_SINGLE)
3307 914 : CopyMultiInsertInfoCleanup(&multiInsertInfo);
3308 :
3309 994 : ExecCloseIndices(target_resultRelInfo);
3310 :
3311 : /* Close all the partitioned tables, leaf partitions, and their indices */
3312 994 : if (proute)
3313 44 : ExecCleanupTupleRouting(mtstate, proute);
3314 :
3315 : /* Close any trigger target relations */
3316 994 : ExecCleanUpTriggerState(estate);
3317 :
3318 994 : FreeExecutorState(estate);
3319 :
3320 994 : return processed;
3321 : }
3322 :
3323 : /*
3324 : * Setup to read tuples from a file for COPY FROM.
3325 : *
3326 : * 'rel': Used as a template for the tuples
3327 : * 'filename': Name of server-local file to read
3328 : * 'attnamelist': List of char *, columns to include. NIL selects all cols.
3329 : * 'options': List of DefElem. See copy_opt_item in gram.y for selections.
3330 : *
3331 : * Returns a CopyState, to be passed to NextCopyFrom and related functions.
3332 : */
3333 : CopyState
3334 1140 : BeginCopyFrom(ParseState *pstate,
3335 : Relation rel,
3336 : const char *filename,
3337 : bool is_program,
3338 : copy_data_source_cb data_source_cb,
3339 : List *attnamelist,
3340 : List *options)
3341 : {
3342 : CopyState cstate;
3343 1140 : bool pipe = (filename == NULL);
3344 : TupleDesc tupDesc;
3345 : AttrNumber num_phys_attrs,
3346 : num_defaults;
3347 : FmgrInfo *in_functions;
3348 : Oid *typioparams;
3349 : int attnum;
3350 : Oid in_func_oid;
3351 : int *defmap;
3352 : ExprState **defexprs;
3353 : MemoryContext oldcontext;
3354 : bool volatile_defexprs;
3355 :
3356 1140 : cstate = BeginCopy(pstate, true, rel, NULL, InvalidOid, attnamelist, options);
3357 1130 : oldcontext = MemoryContextSwitchTo(cstate->copycontext);
3358 :
3359 : /* Initialize state variables */
3360 1130 : cstate->reached_eof = false;
3361 1130 : cstate->eol_type = EOL_UNKNOWN;
3362 1130 : cstate->cur_relname = RelationGetRelationName(cstate->rel);
3363 1130 : cstate->cur_lineno = 0;
3364 1130 : cstate->cur_attname = NULL;
3365 1130 : cstate->cur_attval = NULL;
3366 :
3367 : /* Set up variables to avoid per-attribute overhead. */
3368 1130 : initStringInfo(&cstate->attribute_buf);
3369 1130 : initStringInfo(&cstate->line_buf);
3370 1130 : cstate->line_buf_converted = false;
3371 1130 : cstate->raw_buf = (char *) palloc(RAW_BUF_SIZE + 1);
3372 1130 : cstate->raw_buf_index = cstate->raw_buf_len = 0;
3373 :
3374 : /* Assign range table, we'll need it in CopyFrom. */
3375 1130 : if (pstate)
3376 1076 : cstate->range_table = pstate->p_rtable;
3377 :
3378 1130 : tupDesc = RelationGetDescr(cstate->rel);
3379 1130 : num_phys_attrs = tupDesc->natts;
3380 1130 : num_defaults = 0;
3381 1130 : volatile_defexprs = false;
3382 :
3383 : /*
3384 : * Pick up the required catalog information for each attribute in the
3385 : * relation, including the input function, the element type (to pass to
3386 : * the input function), and info about defaults and constraints. (Which
3387 : * input function we use depends on text/binary format choice.)
3388 : */
3389 1130 : in_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
3390 1130 : typioparams = (Oid *) palloc(num_phys_attrs * sizeof(Oid));
3391 1130 : defmap = (int *) palloc(num_phys_attrs * sizeof(int));
3392 1130 : defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));
3393 :
3394 5798 : for (attnum = 1; attnum <= num_phys_attrs; attnum++)
3395 : {
3396 4668 : Form_pg_attribute att = TupleDescAttr(tupDesc, attnum - 1);
3397 :
3398 : /* We don't need info for dropped attributes */
3399 4668 : if (att->attisdropped)
3400 16 : continue;
3401 :
3402 : /* Fetch the input function and typioparam info */
3403 4652 : if (cstate->binary)
3404 28 : getTypeBinaryInputInfo(att->atttypid,
3405 28 : &in_func_oid, &typioparams[attnum - 1]);
3406 : else
3407 4624 : getTypeInputInfo(att->atttypid,
3408 4624 : &in_func_oid, &typioparams[attnum - 1]);
3409 4652 : fmgr_info(in_func_oid, &in_functions[attnum - 1]);
3410 :
3411 : /* Get default info if needed */
3412 4652 : if (!list_member_int(cstate->attnumlist, attnum) && !att->attgenerated)
3413 : {
3414 : /* attribute is NOT to be copied from input */
3415 : /* use default value if one exists */
3416 508 : Expr *defexpr = (Expr *) build_column_default(cstate->rel,
3417 : attnum);
3418 :
3419 508 : if (defexpr != NULL)
3420 : {
3421 : /* Run the expression through planner */
3422 82 : defexpr = expression_planner(defexpr);
3423 :
3424 : /* Initialize executable expression in copycontext */
3425 82 : defexprs[num_defaults] = ExecInitExpr(defexpr, NULL);
3426 82 : defmap[num_defaults] = attnum - 1;
3427 82 : num_defaults++;
3428 :
3429 : /*
3430 : * If a default expression looks at the table being loaded,
3431 : * then it could give the wrong answer when using
3432 : * multi-insert. Since database access can be dynamic this is
3433 : * hard to test for exactly, so we use the much wider test of
3434 : * whether the default expression is volatile. We allow for
3435 : * the special case of when the default expression is the
3436 : * nextval() of a sequence which in this specific case is
3437 : * known to be safe for use with the multi-insert
3438 : * optimization. Hence we use this special case function
3439 : * checker rather than the standard check for
3440 : * contain_volatile_functions().
3441 : */
3442 82 : if (!volatile_defexprs)
3443 82 : volatile_defexprs = contain_volatile_functions_not_nextval((Node *) defexpr);
3444 : }
3445 : }
3446 : }
3447 :
3448 : /* We keep those variables in cstate. */
3449 1130 : cstate->in_functions = in_functions;
3450 1130 : cstate->typioparams = typioparams;
3451 1130 : cstate->defmap = defmap;
3452 1130 : cstate->defexprs = defexprs;
3453 1130 : cstate->volatile_defexprs = volatile_defexprs;
3454 1130 : cstate->num_defaults = num_defaults;
3455 1130 : cstate->is_program = is_program;
3456 :
3457 1130 : if (data_source_cb)
3458 : {
3459 102 : cstate->copy_dest = COPY_CALLBACK;
3460 102 : cstate->data_source_cb = data_source_cb;
3461 : }
3462 1028 : else if (pipe)
3463 : {
3464 : Assert(!is_program); /* the grammar does not allow this */
3465 472 : if (whereToSendOutput == DestRemote)
3466 472 : ReceiveCopyBegin(cstate);
3467 : else
3468 0 : cstate->copy_file = stdin;
3469 : }
3470 : else
3471 : {
3472 556 : cstate->filename = pstrdup(filename);
3473 :
3474 556 : if (cstate->is_program)
3475 : {
3476 0 : cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_R);
3477 0 : if (cstate->copy_file == NULL)
3478 0 : ereport(ERROR,
3479 : (errcode_for_file_access(),
3480 : errmsg("could not execute command \"%s\": %m",
3481 : cstate->filename)));
3482 : }
3483 : else
3484 : {
3485 : struct stat st;
3486 :
3487 556 : cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_R);
3488 556 : if (cstate->copy_file == NULL)
3489 : {
3490 : /* copy errno because ereport subfunctions might change it */
3491 0 : int save_errno = errno;
3492 :
3493 0 : ereport(ERROR,
3494 : (errcode_for_file_access(),
3495 : errmsg("could not open file \"%s\" for reading: %m",
3496 : cstate->filename),
3497 : (save_errno == ENOENT || save_errno == EACCES) ?
3498 : errhint("COPY FROM instructs the PostgreSQL server process to read a file. "
3499 : "You may want a client-side facility such as psql's \\copy.") : 0));
3500 : }
3501 :
3502 556 : if (fstat(fileno(cstate->copy_file), &st))
3503 0 : ereport(ERROR,
3504 : (errcode_for_file_access(),
3505 : errmsg("could not stat file \"%s\": %m",
3506 : cstate->filename)));
3507 :
3508 556 : if (S_ISDIR(st.st_mode))
3509 0 : ereport(ERROR,
3510 : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
3511 : errmsg("\"%s\" is a directory", cstate->filename)));
3512 : }
3513 : }
3514 :
3515 1130 : if (cstate->binary)
3516 : {
3517 : /* Read and verify binary header */
3518 : char readSig[11];
3519 : int32 tmp;
3520 :
3521 : /* Signature */
3522 4 : if (CopyGetData(cstate, readSig, 11, 11) != 11 ||
3523 4 : memcmp(readSig, BinarySignature, 11) != 0)
3524 0 : ereport(ERROR,
3525 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3526 : errmsg("COPY file signature not recognized")));
3527 : /* Flags field */
3528 4 : if (!CopyGetInt32(cstate, &tmp))
3529 0 : ereport(ERROR,
3530 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3531 : errmsg("invalid COPY file header (missing flags)")));
3532 4 : if ((tmp & (1 << 16)) != 0)
3533 0 : ereport(ERROR,
3534 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3535 : errmsg("invalid COPY file header (WITH OIDS)")));
3536 4 : tmp &= ~(1 << 16);
3537 4 : if ((tmp >> 16) != 0)
3538 0 : ereport(ERROR,
3539 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3540 : errmsg("unrecognized critical flags in COPY file header")));
3541 : /* Header extension length */
3542 4 : if (!CopyGetInt32(cstate, &tmp) ||
3543 4 : tmp < 0)
3544 0 : ereport(ERROR,
3545 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3546 : errmsg("invalid COPY file header (missing length)")));
3547 : /* Skip extension header, if present */
3548 4 : while (tmp-- > 0)
3549 : {
3550 0 : if (CopyGetData(cstate, readSig, 1, 1) != 1)
3551 0 : ereport(ERROR,
3552 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3553 : errmsg("invalid COPY file header (wrong length)")));
3554 : }
3555 : }
3556 :
3557 : /* create workspace for CopyReadAttributes results */
3558 1130 : if (!cstate->binary)
3559 : {
3560 1126 : AttrNumber attr_count = list_length(cstate->attnumlist);
3561 :
3562 1126 : cstate->max_fields = attr_count;
3563 1126 : cstate->raw_fields = (char **) palloc(attr_count * sizeof(char *));
3564 : }
3565 :
3566 1130 : MemoryContextSwitchTo(oldcontext);
3567 :
3568 1130 : return cstate;
3569 : }
3570 :
3571 : /*
3572 : * Read raw fields in the next line for COPY FROM in text or csv mode.
3573 : * Return false if no more lines.
3574 : *
3575 : * An internal temporary buffer is returned via 'fields'. It is valid until
3576 : * the next call of the function. Since the function returns all raw fields
3577 : * in the input file, 'nfields' could be different from the number of columns
3578 : * in the relation.
3579 : *
3580 : * NOTE: force_not_null option are not applied to the returned fields.
3581 : */
3582 : bool
3583 1157068 : NextCopyFromRawFields(CopyState cstate, char ***fields, int *nfields)
3584 : {
3585 : int fldct;
3586 : bool done;
3587 :
3588 : /* only available for text or csv input */
3589 : Assert(!cstate->binary);
3590 :
3591 : /* on input just throw the header line away */
3592 1157068 : if (cstate->cur_lineno == 0 && cstate->header_line)
3593 : {
3594 26 : cstate->cur_lineno++;
3595 26 : if (CopyReadLine(cstate))
3596 0 : return false; /* done */
3597 : }
3598 :
3599 1157068 : cstate->cur_lineno++;
3600 :
3601 : /* Actually read the line into memory here */
3602 1157068 : done = CopyReadLine(cstate);
3603 :
3604 : /*
3605 : * EOF at start of line means we're done. If we see EOF after some
3606 : * characters, we act as though it was newline followed by EOF, ie,
3607 : * process the line and then exit loop on next iteration.
3608 : */
3609 1157068 : if (done && cstate->line_buf.len == 0)
3610 1044 : return false;
3611 :
3612 : /* Parse the line into de-escaped field values */
3613 1156024 : if (cstate->csv_mode)
3614 238 : fldct = CopyReadAttributesCSV(cstate);
3615 : else
3616 1155786 : fldct = CopyReadAttributesText(cstate);
3617 :
3618 1156024 : *fields = cstate->raw_fields;
3619 1156024 : *nfields = fldct;
3620 1156024 : return true;
3621 : }
3622 :
3623 : /*
3624 : * Read next tuple from file for COPY FROM. Return false if no more tuples.
3625 : *
3626 : * 'econtext' is used to evaluate default expression for each columns not
3627 : * read from the file. It can be NULL when no default values are used, i.e.
3628 : * when all columns are read from the file.
3629 : *
3630 : * 'values' and 'nulls' arrays must be the same length as columns of the
3631 : * relation passed to BeginCopyFrom. This function fills the arrays.
3632 : * Oid of the tuple is returned with 'tupleOid' separately.
3633 : */
3634 : bool
3635 1157084 : NextCopyFrom(CopyState cstate, ExprContext *econtext,
3636 : Datum *values, bool *nulls)
3637 : {
3638 : TupleDesc tupDesc;
3639 : AttrNumber num_phys_attrs,
3640 : attr_count,
3641 1157084 : num_defaults = cstate->num_defaults;
3642 1157084 : FmgrInfo *in_functions = cstate->in_functions;
3643 1157084 : Oid *typioparams = cstate->typioparams;
3644 : int i;
3645 1157084 : int *defmap = cstate->defmap;
3646 1157084 : ExprState **defexprs = cstate->defexprs;
3647 :
3648 1157084 : tupDesc = RelationGetDescr(cstate->rel);
3649 1157084 : num_phys_attrs = tupDesc->natts;
3650 1157084 : attr_count = list_length(cstate->attnumlist);
3651 :
3652 : /* Initialize all values for row to NULL */
3653 6257178 : MemSet(values, 0, num_phys_attrs * sizeof(Datum));
3654 1157084 : MemSet(nulls, true, num_phys_attrs * sizeof(bool));
3655 :
3656 1157084 : if (!cstate->binary)
3657 : {
3658 : char **field_strings;
3659 : ListCell *cur;
3660 : int fldct;
3661 : int fieldno;
3662 : char *string;
3663 :
3664 : /* read raw fields in the next line */
3665 1157068 : if (!NextCopyFromRawFields(cstate, &field_strings, &fldct))
3666 1044 : return false;
3667 :
3668 : /* check for overflowing fields */
3669 1156024 : if (attr_count > 0 && fldct > attr_count)
3670 8 : ereport(ERROR,
3671 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3672 : errmsg("extra data after last expected column")));
3673 :
3674 1156016 : fieldno = 0;
3675 :
3676 : /* Loop to read the user attributes on the line. */
3677 5996344 : foreach(cur, cstate->attnumlist)
3678 : {
3679 4840354 : int attnum = lfirst_int(cur);
3680 4840354 : int m = attnum - 1;
3681 4840354 : Form_pg_attribute att = TupleDescAttr(tupDesc, m);
3682 :
3683 4840354 : if (fieldno >= fldct)
3684 8 : ereport(ERROR,
3685 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3686 : errmsg("missing data for column \"%s\"",
3687 : NameStr(att->attname))));
3688 4840346 : string = field_strings[fieldno++];
3689 :
3690 4840346 : if (cstate->convert_select_flags &&
3691 20 : !cstate->convert_select_flags[m])
3692 : {
3693 : /* ignore input field, leaving column as NULL */
3694 10 : continue;
3695 : }
3696 :
3697 4840336 : if (cstate->csv_mode)
3698 : {
3699 474 : if (string == NULL &&
3700 26 : cstate->force_notnull_flags[m])
3701 : {
3702 : /*
3703 : * FORCE_NOT_NULL option is set and column is NULL -
3704 : * convert it to the NULL string.
3705 : */
3706 12 : string = cstate->null_print;
3707 : }
3708 462 : else if (string != NULL && cstate->force_null_flags[m]
3709 16 : && strcmp(string, cstate->null_print) == 0)
3710 : {
3711 : /*
3712 : * FORCE_NULL option is set and column matches the NULL
3713 : * string. It must have been quoted, or otherwise the
3714 : * string would already have been set to NULL. Convert it
3715 : * to NULL as specified.
3716 : */
3717 10 : string = NULL;
3718 : }
3719 : }
3720 :
3721 4840336 : cstate->cur_attname = NameStr(att->attname);
3722 4840336 : cstate->cur_attval = string;
3723 14520990 : values[m] = InputFunctionCall(&in_functions[m],
3724 : string,
3725 4840336 : typioparams[m],
3726 : att->atttypmod);
3727 4840318 : if (string != NULL)
3728 4836202 : nulls[m] = false;
3729 4840318 : cstate->cur_attname = NULL;
3730 4840318 : cstate->cur_attval = NULL;
3731 : }
3732 :
3733 : Assert(fieldno == attr_count);
3734 : }
3735 : else
3736 : {
3737 : /* binary */
3738 : int16 fld_count;
3739 : ListCell *cur;
3740 :
3741 16 : cstate->cur_lineno++;
3742 :
3743 16 : if (!CopyGetInt16(cstate, &fld_count))
3744 : {
3745 : /* EOF detected (end of file, or protocol-level EOF) */
3746 4 : return false;
3747 : }
3748 :
3749 16 : if (fld_count == -1)
3750 : {
3751 : /*
3752 : * Received EOF marker. In a V3-protocol copy, wait for the
3753 : * protocol-level EOF, and complain if it doesn't come
3754 : * immediately. This ensures that we correctly handle CopyFail,
3755 : * if client chooses to send that now.
3756 : *
3757 : * Note that we MUST NOT try to read more data in an old-protocol
3758 : * copy, since there is no protocol-level EOF marker then. We
3759 : * could go either way for copy from file, but choose to throw
3760 : * error if there's data after the EOF marker, for consistency
3761 : * with the new-protocol case.
3762 : */
3763 : char dummy;
3764 :
3765 8 : if (cstate->copy_dest != COPY_OLD_FE &&
3766 4 : CopyGetData(cstate, &dummy, 1, 1) > 0)
3767 0 : ereport(ERROR,
3768 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3769 : errmsg("received copy data after EOF marker")));
3770 4 : return false;
3771 : }
3772 :
3773 12 : if (fld_count != attr_count)
3774 0 : ereport(ERROR,
3775 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
3776 : errmsg("row field count is %d, expected %d",
3777 : (int) fld_count, attr_count)));
3778 :
3779 12 : i = 0;
3780 96 : foreach(cur, cstate->attnumlist)
3781 : {
3782 84 : int attnum = lfirst_int(cur);
3783 84 : int m = attnum - 1;
3784 84 : Form_pg_attribute att = TupleDescAttr(tupDesc, m);
3785 :
3786 84 : cstate->cur_attname = NameStr(att->attname);
3787 84 : i++;
3788 336 : values[m] = CopyReadBinaryAttribute(cstate,
3789 : i,
3790 84 : &in_functions[m],
3791 84 : typioparams[m],
3792 : att->atttypmod,
3793 : &nulls[m]);
3794 84 : cstate->cur_attname = NULL;
3795 : }
3796 : }
3797 :
3798 : /*
3799 : * Now compute and insert any defaults available for the columns not
3800 : * provided by the input data. Anything not processed here or above will
3801 : * remain NULL.
3802 : */
3803 1156104 : for (i = 0; i < num_defaults; i++)
3804 : {
3805 : /*
3806 : * The caller must supply econtext and have switched into the
3807 : * per-tuple memory context in it.
3808 : */
3809 : Assert(econtext != NULL);
3810 : Assert(CurrentMemoryContext == econtext->ecxt_per_tuple_memory);
3811 :
3812 102 : values[defmap[i]] = ExecEvalExpr(defexprs[i], econtext,
3813 102 : &nulls[defmap[i]]);
3814 : }
3815 :
3816 1156002 : return true;
3817 : }
3818 :
3819 : /*
3820 : * Clean up storage and release resources for COPY FROM.
3821 : */
3822 : void
3823 948 : EndCopyFrom(CopyState cstate)
3824 : {
3825 : /* No COPY FROM related resources except memory. */
3826 :
3827 948 : EndCopy(cstate);
3828 948 : }
3829 :
3830 : /*
3831 : * Read the next input line and stash it in line_buf, with conversion to
3832 : * server encoding.
3833 : *
3834 : * Result is true if read was terminated by EOF, false if terminated
3835 : * by newline. The terminating newline or EOF marker is not included
3836 : * in the final value of line_buf.
3837 : */
3838 : static bool
3839 1157094 : CopyReadLine(CopyState cstate)
3840 : {
3841 : bool result;
3842 :
3843 1157094 : resetStringInfo(&cstate->line_buf);
3844 1157094 : cstate->line_buf_valid = true;
3845 :
3846 : /* Mark that encoding conversion hasn't occurred yet */
3847 1157094 : cstate->line_buf_converted = false;
3848 :
3849 : /* Parse data and transfer into line_buf */
3850 1157094 : result = CopyReadLineText(cstate);
3851 :
3852 1157094 : if (result)
3853 : {
3854 : /*
3855 : * Reached EOF. In protocol version 3, we should ignore anything
3856 : * after \. up to the protocol end of copy data. (XXX maybe better
3857 : * not to treat \. as special?)
3858 : */
3859 1044 : if (cstate->copy_dest == COPY_NEW_FE)
3860 : {
3861 : do
3862 : {
3863 404 : cstate->raw_buf_index = cstate->raw_buf_len;
3864 404 : } while (CopyLoadRawBuf(cstate));
3865 : }
3866 : }
3867 : else
3868 : {
3869 : /*
3870 : * If we didn't hit EOF, then we must have transferred the EOL marker
3871 : * to line_buf along with the data. Get rid of it.
3872 : */
3873 1156050 : switch (cstate->eol_type)
3874 : {
3875 1156050 : case EOL_NL:
3876 : Assert(cstate->line_buf.len >= 1);
3877 : Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\n');
3878 1156050 : cstate->line_buf.len--;
3879 1156050 : cstate->line_buf.data[cstate->line_buf.len] = '\0';
3880 1156050 : break;
3881 0 : case EOL_CR:
3882 : Assert(cstate->line_buf.len >= 1);
3883 : Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\r');
3884 0 : cstate->line_buf.len--;
3885 0 : cstate->line_buf.data[cstate->line_buf.len] = '\0';
3886 0 : break;
3887 0 : case EOL_CRNL:
3888 : Assert(cstate->line_buf.len >= 2);
3889 : Assert(cstate->line_buf.data[cstate->line_buf.len - 2] == '\r');
3890 : Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\n');
3891 0 : cstate->line_buf.len -= 2;
3892 0 : cstate->line_buf.data[cstate->line_buf.len] = '\0';
3893 0 : break;
3894 0 : case EOL_UNKNOWN:
3895 : /* shouldn't get here */
3896 : Assert(false);
3897 0 : break;
3898 : }
3899 1157094 : }
3900 :
3901 : /* Done reading the line. Convert it to server encoding. */
3902 1157094 : if (cstate->need_transcoding)
3903 : {
3904 : char *cvt;
3905 :
3906 1148524 : cvt = pg_any_to_server(cstate->line_buf.data,
3907 : cstate->line_buf.len,
3908 : cstate->file_encoding);
3909 1148524 : if (cvt != cstate->line_buf.data)
3910 : {
3911 : /* transfer converted data back to line_buf */
3912 0 : resetStringInfo(&cstate->line_buf);
3913 0 : appendBinaryStringInfo(&cstate->line_buf, cvt, strlen(cvt));
3914 0 : pfree(cvt);
3915 : }
3916 : }
3917 :
3918 : /* Now it's safe to use the buffer in error messages */
3919 1157094 : cstate->line_buf_converted = true;
3920 :
3921 1157094 : return result;
3922 : }
3923 :
3924 : /*
3925 : * CopyReadLineText - inner loop of CopyReadLine for text mode
3926 : */
3927 : static bool
3928 1157094 : CopyReadLineText(CopyState cstate)
3929 : {
3930 : char *copy_raw_buf;
3931 : int raw_buf_ptr;
3932 : int copy_buf_len;
3933 1157094 : bool need_data = false;
3934 1157094 : bool hit_eof = false;
3935 1157094 : bool result = false;
3936 : char mblen_str[2];
3937 :
3938 : /* CSV variables */
3939 1157094 : bool first_char_in_line = true;
3940 1157094 : bool in_quote = false,
3941 1157094 : last_was_esc = false;
3942 1157094 : char quotec = '\0';
3943 1157094 : char escapec = '\0';
3944 :
3945 1157094 : if (cstate->csv_mode)
3946 : {
3947 358 : quotec = cstate->quote[0];
3948 358 : escapec = cstate->escape[0];
3949 : /* ignore special escape processing if it's the same as quotec */
3950 358 : if (quotec == escapec)
3951 232 : escapec = '\0';
3952 : }
3953 :
3954 1157094 : mblen_str[1] = '\0';
3955 :
3956 : /*
3957 : * The objective of this loop is to transfer the entire next input line
3958 : * into line_buf. Hence, we only care for detecting newlines (\r and/or
3959 : * \n) and the end-of-copy marker (\.).
3960 : *
3961 : * In CSV mode, \r and \n inside a quoted field are just part of the data
3962 : * value and are put in line_buf. We keep just enough state to know if we
3963 : * are currently in a quoted field or not.
3964 : *
3965 : * These four characters, and the CSV escape and quote characters, are
3966 : * assumed the same in frontend and backend encodings.
3967 : *
3968 : * For speed, we try to move data from raw_buf to line_buf in chunks
3969 : * rather than one character at a time. raw_buf_ptr points to the next
3970 : * character to examine; any characters from raw_buf_index to raw_buf_ptr
3971 : * have been determined to be part of the line, but not yet transferred to
3972 : * line_buf.
3973 : *
3974 : * For a little extra speed within the loop, we copy raw_buf and
3975 : * raw_buf_len into local variables.
3976 : */
3977 1157094 : copy_raw_buf = cstate->raw_buf;
3978 1157094 : raw_buf_ptr = cstate->raw_buf_index;
3979 1157094 : copy_buf_len = cstate->raw_buf_len;
3980 :
3981 : for (;;)
3982 28619418 : {
3983 : int prev_raw_ptr;
3984 : char c;
3985 :
3986 : /*
3987 : * Load more data if needed. Ideally we would just force four bytes
3988 : * of read-ahead and avoid the many calls to
3989 : * IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(), but the COPY_OLD_FE protocol
3990 : * does not allow us to read too far ahead or we might read into the
3991 : * next data, so we read-ahead only as far we know we can. One
3992 : * optimization would be to read-ahead four byte here if
3993 : * cstate->copy_dest != COPY_OLD_FE, but it hardly seems worth it,
3994 : * considering the size of the buffer.
3995 : */
3996 29776512 : if (raw_buf_ptr >= copy_buf_len || need_data)
3997 : {
3998 477304 : REFILL_LINEBUF;
3999 :
4000 : /*
4001 : * Try to read some more data. This will certainly reset
4002 : * raw_buf_index to zero, and raw_buf_ptr must go with it.
4003 : */
4004 477304 : if (!CopyLoadRawBuf(cstate))
4005 1036 : hit_eof = true;
4006 477304 : raw_buf_ptr = 0;
4007 477304 : copy_buf_len = cstate->raw_buf_len;
4008 :
4009 : /*
4010 : * If we are completely out of data, break out of the loop,
4011 : * reporting EOF.
4012 : */
4013 477304 : if (copy_buf_len <= 0)
4014 : {
4015 1036 : result = true;
4016 1036 : break;
4017 : }
4018 476268 : need_data = false;
4019 : }
4020 :
4021 : /* OK to fetch a character */
4022 29775476 : prev_raw_ptr = raw_buf_ptr;
4023 29775476 : c = copy_raw_buf[raw_buf_ptr++];
4024 :
4025 29775476 : if (cstate->csv_mode)
4026 : {
4027 : /*
4028 : * If character is '\\' or '\r', we may need to look ahead below.
4029 : * Force fetch of the next character if we don't already have it.
4030 : * We need to do this before changing CSV state, in case one of
4031 : * these characters is also the quote or escape character.
4032 : *
4033 : * Note: old-protocol does not like forced prefetch, but it's OK
4034 : * here since we cannot validly be at EOF.
4035 : */
4036 2708 : if (c == '\\' || c == '\r')
4037 : {
4038 96 : IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(0);
4039 : }
4040 :
4041 : /*
4042 : * Dealing with quotes and escapes here is mildly tricky. If the
4043 : * quote char is also the escape char, there's no problem - we
4044 : * just use the char as a toggle. If they are different, we need
4045 : * to ensure that we only take account of an escape inside a
4046 : * quoted field and immediately preceding a quote char, and not
4047 : * the second in an escape-escape sequence.
4048 : */
4049 2708 : if (in_quote && c == escapec)
4050 32 : last_was_esc = !last_was_esc;
4051 2708 : if (c == quotec && !last_was_esc)
4052 296 : in_quote = !in_quote;
4053 2708 : if (c != escapec)
4054 2672 : last_was_esc = false;
4055 :
4056 : /*
4057 : * Updating the line count for embedded CR and/or LF chars is
4058 : * necessarily a little fragile - this test is probably about the
4059 : * best we can do. (XXX it's arguable whether we should do this
4060 : * at all --- is cur_lineno a physical or logical count?)
4061 : */
4062 2708 : if (in_quote && c == (cstate->eol_type == EOL_NL ? '\n' : '\r'))
4063 24 : cstate->cur_lineno++;
4064 : }
4065 :
4066 : /* Process \r */
4067 29775476 : if (c == '\r' && (!cstate->csv_mode || !in_quote))
4068 : {
4069 : /* Check for \r\n on first line, _and_ handle \r\n. */
4070 0 : if (cstate->eol_type == EOL_UNKNOWN ||
4071 0 : cstate->eol_type == EOL_CRNL)
4072 : {
4073 : /*
4074 : * If need more data, go back to loop top to load it.
4075 : *
4076 : * Note that if we are at EOF, c will wind up as '\0' because
4077 : * of the guaranteed pad of raw_buf.
4078 : */
4079 0 : IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(0);
4080 :
4081 : /* get next char */
4082 0 : c = copy_raw_buf[raw_buf_ptr];
4083 :
4084 0 : if (c == '\n')
4085 : {
4086 0 : raw_buf_ptr++; /* eat newline */
4087 0 : cstate->eol_type = EOL_CRNL; /* in case not set yet */
4088 : }
4089 : else
4090 : {
4091 : /* found \r, but no \n */
4092 0 : if (cstate->eol_type == EOL_CRNL)
4093 0 : ereport(ERROR,
4094 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4095 : !cstate->csv_mode ?
4096 : errmsg("literal carriage return found in data") :
4097 : errmsg("unquoted carriage return found in data"),
4098 : !cstate->csv_mode ?
4099 : errhint("Use \"\\r\" to represent carriage return.") :
4100 : errhint("Use quoted CSV field to represent carriage return.")));
4101 :
4102 : /*
4103 : * if we got here, it is the first line and we didn't find
4104 : * \n, so don't consume the peeked character
4105 : */
4106 0 : cstate->eol_type = EOL_CR;
4107 : }
4108 : }
4109 0 : else if (cstate->eol_type == EOL_NL)
4110 0 : ereport(ERROR,
4111 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4112 : !cstate->csv_mode ?
4113 : errmsg("literal carriage return found in data") :
4114 : errmsg("unquoted carriage return found in data"),
4115 : !cstate->csv_mode ?
4116 : errhint("Use \"\\r\" to represent carriage return.") :
4117 : errhint("Use quoted CSV field to represent carriage return.")));
4118 : /* If reach here, we have found the line terminator */
4119 0 : break;
4120 : }
4121 :
4122 : /* Process \n */
4123 29775476 : if (c == '\n' && (!cstate->csv_mode || !in_quote))
4124 : {
4125 1156050 : if (cstate->eol_type == EOL_CR || cstate->eol_type == EOL_CRNL)
4126 0 : ereport(ERROR,
4127 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4128 : !cstate->csv_mode ?
4129 : errmsg("literal newline found in data") :
4130 : errmsg("unquoted newline found in data"),
4131 : !cstate->csv_mode ?
4132 : errhint("Use \"\\n\" to represent newline.") :
4133 : errhint("Use quoted CSV field to represent newline.")));
4134 1156050 : cstate->eol_type = EOL_NL; /* in case not set yet */
4135 : /* If reach here, we have found the line terminator */
4136 1156050 : break;
4137 : }
4138 :
4139 : /*
4140 : * In CSV mode, we only recognize \. alone on a line. This is because
4141 : * \. is a valid CSV data value.
4142 : */
4143 28619426 : if (c == '\\' && (!cstate->csv_mode || first_char_in_line))
4144 : {
4145 : char c2;
4146 :
4147 6182 : IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(0);
4148 6182 : IF_NEED_REFILL_AND_EOF_BREAK(0);
4149 :
4150 : /* -----
4151 : * get next character
4152 : * Note: we do not change c so if it isn't \., we can fall
4153 : * through and continue processing for file encoding.
4154 : * -----
4155 : */
4156 6182 : c2 = copy_raw_buf[raw_buf_ptr];
4157 :
4158 6182 : if (c2 == '.')
4159 : {
4160 12 : raw_buf_ptr++; /* consume the '.' */
4161 :
4162 : /*
4163 : * Note: if we loop back for more data here, it does not
4164 : * matter that the CSV state change checks are re-executed; we
4165 : * will come back here with no important state changed.
4166 : */
4167 12 : if (cstate->eol_type == EOL_CRNL)
4168 : {
4169 : /* Get the next character */
4170 0 : IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(0);
4171 : /* if hit_eof, c2 will become '\0' */
4172 0 : c2 = copy_raw_buf[raw_buf_ptr++];
4173 :
4174 0 : if (c2 == '\n')
4175 : {
4176 0 : if (!cstate->csv_mode)
4177 0 : ereport(ERROR,
4178 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4179 : errmsg("end-of-copy marker does not match previous newline style")));
4180 : else
4181 0 : NO_END_OF_COPY_GOTO;
4182 : }
4183 0 : else if (c2 != '\r')
4184 : {
4185 0 : if (!cstate->csv_mode)
4186 0 : ereport(ERROR,
4187 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4188 : errmsg("end-of-copy marker corrupt")));
4189 : else
4190 0 : NO_END_OF_COPY_GOTO;
4191 : }
4192 : }
4193 :
4194 : /* Get the next character */
4195 12 : IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(0);
4196 : /* if hit_eof, c2 will become '\0' */
4197 12 : c2 = copy_raw_buf[raw_buf_ptr++];
4198 :
4199 12 : if (c2 != '\r' && c2 != '\n')
4200 : {
4201 4 : if (!cstate->csv_mode)
4202 0 : ereport(ERROR,
4203 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4204 : errmsg("end-of-copy marker corrupt")));
4205 : else
4206 4 : NO_END_OF_COPY_GOTO;
4207 : }
4208 :
4209 8 : if ((cstate->eol_type == EOL_NL && c2 != '\n') ||
4210 8 : (cstate->eol_type == EOL_CRNL && c2 != '\n') ||
4211 8 : (cstate->eol_type == EOL_CR && c2 != '\r'))
4212 : {
4213 0 : ereport(ERROR,
4214 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4215 : errmsg("end-of-copy marker does not match previous newline style")));
4216 : }
4217 :
4218 : /*
4219 : * Transfer only the data before the \. into line_buf, then
4220 : * discard the data and the \. sequence.
4221 : */
4222 8 : if (prev_raw_ptr > cstate->raw_buf_index)
4223 0 : appendBinaryStringInfo(&cstate->line_buf,
4224 0 : cstate->raw_buf + cstate->raw_buf_index,
4225 0 : prev_raw_ptr - cstate->raw_buf_index);
4226 8 : cstate->raw_buf_index = raw_buf_ptr;
4227 8 : result = true; /* report EOF */
4228 8 : break;
4229 : }
4230 6170 : else if (!cstate->csv_mode)
4231 :
4232 : /*
4233 : * If we are here, it means we found a backslash followed by
4234 : * something other than a period. In non-CSV mode, anything
4235 : * after a backslash is special, so we skip over that second
4236 : * character too. If we didn't do that \\. would be
4237 : * considered an eof-of copy, while in non-CSV mode it is a
4238 : * literal backslash followed by a period. In CSV mode,
4239 : * backslashes are not special, so we want to process the
4240 : * character after the backslash just like a normal character,
4241 : * so we don't increment in those cases.
4242 : */
4243 6170 : raw_buf_ptr++;
4244 : }
4245 :
4246 : /*
4247 : * This label is for CSV cases where \. appears at the start of a
4248 : * line, but there is more text after it, meaning it was a data value.
4249 : * We are more strict for \. in CSV mode because \. could be a data
4250 : * value, while in non-CSV mode, \. cannot be a data value.
4251 : */
4252 28613244 : not_end_of_copy:
4253 :
4254 : /*
4255 : * Process all bytes of a multi-byte character as a group.
4256 : *
4257 : * We only support multi-byte sequences where the first byte has the
4258 : * high-bit set, so as an optimization we can avoid this block
4259 : * entirely if it is not set.
4260 : */
4261 28619418 : if (cstate->encoding_embeds_ascii && IS_HIGHBIT_SET(c))
4262 : {
4263 : int mblen;
4264 :
4265 : /*
4266 : * It is enough to look at the first byte in all our encodings, to
4267 : * get the length. (GB18030 is a bit special, but still works for
4268 : * our purposes; see comment in pg_gb18030_mblen())
4269 : */
4270 0 : mblen_str[0] = c;
4271 0 : mblen = pg_encoding_mblen(cstate->file_encoding, mblen_str);
4272 :
4273 0 : IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(mblen - 1);
4274 0 : IF_NEED_REFILL_AND_EOF_BREAK(mblen - 1);
4275 0 : raw_buf_ptr += mblen - 1;
4276 : }
4277 28619418 : first_char_in_line = false;
4278 : } /* end of outer loop */
4279 :
4280 : /*
4281 : * Transfer any still-uncopied data to line_buf.
4282 : */
4283 1157094 : REFILL_LINEBUF;
4284 :
4285 1157094 : return result;
4286 : }
4287 :
4288 : /*
4289 : * Return decimal value for a hexadecimal digit
4290 : */
4291 : static int
4292 0 : GetDecimalFromHex(char hex)
4293 : {
4294 0 : if (isdigit((unsigned char) hex))
4295 0 : return hex - '0';
4296 : else
4297 0 : return tolower((unsigned char) hex) - 'a' + 10;
4298 : }
4299 :
4300 : /*
4301 : * Parse the current line into separate attributes (fields),
4302 : * performing de-escaping as needed.
4303 : *
4304 : * The input is in line_buf. We use attribute_buf to hold the result
4305 : * strings. cstate->raw_fields[k] is set to point to the k'th attribute
4306 : * string, or NULL when the input matches the null marker string.
4307 : * This array is expanded as necessary.
4308 : *
4309 : * (Note that the caller cannot check for nulls since the returned
4310 : * string would be the post-de-escaping equivalent, which may look
4311 : * the same as some valid data string.)
4312 : *
4313 : * delim is the column delimiter string (must be just one byte for now).
4314 : * null_print is the null marker string. Note that this is compared to
4315 : * the pre-de-escaped input string.
4316 : *
4317 : * The return value is the number of fields actually read.
4318 : */
4319 : static int
4320 1155786 : CopyReadAttributesText(CopyState cstate)
4321 : {
4322 1155786 : char delimc = cstate->delim[0];
4323 : int fieldno;
4324 : char *output_ptr;
4325 : char *cur_ptr;
4326 : char *line_end_ptr;
4327 :
4328 : /*
4329 : * We need a special case for zero-column tables: check that the input
4330 : * line is empty, and return.
4331 : */
4332 1155786 : if (cstate->max_fields <= 0)
4333 : {
4334 0 : if (cstate->line_buf.len != 0)
4335 0 : ereport(ERROR,
4336 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4337 : errmsg("extra data after last expected column")));
4338 0 : return 0;
4339 : }
4340 :
4341 1155786 : resetStringInfo(&cstate->attribute_buf);
4342 :
4343 : /*
4344 : * The de-escaped attributes will certainly not be longer than the input
4345 : * data line, so we can just force attribute_buf to be large enough and
4346 : * then transfer data without any checks for enough space. We need to do
4347 : * it this way because enlarging attribute_buf mid-stream would invalidate
4348 : * pointers already stored into cstate->raw_fields[].
4349 : */
4350 1155786 : if (cstate->attribute_buf.maxlen <= cstate->line_buf.len)
4351 6 : enlargeStringInfo(&cstate->attribute_buf, cstate->line_buf.len);
4352 1155786 : output_ptr = cstate->attribute_buf.data;
4353 :
4354 : /* set pointer variables for loop */
4355 1155786 : cur_ptr = cstate->line_buf.data;
4356 1155786 : line_end_ptr = cstate->line_buf.data + cstate->line_buf.len;
4357 :
4358 : /* Outer loop iterates over fields */
4359 1155786 : fieldno = 0;
4360 : for (;;)
4361 3684116 : {
4362 4839902 : bool found_delim = false;
4363 : char *start_ptr;
4364 : char *end_ptr;
4365 : int input_len;
4366 4839902 : bool saw_non_ascii = false;
4367 :
4368 : /* Make sure there is enough space for the next value */
4369 4839902 : if (fieldno >= cstate->max_fields)
4370 : {
4371 8 : cstate->max_fields *= 2;
4372 8 : cstate->raw_fields =
4373 8 : repalloc(cstate->raw_fields, cstate->max_fields * sizeof(char *));
4374 : }
4375 :
4376 : /* Remember start of field on both input and output sides */
4377 4839902 : start_ptr = cur_ptr;
4378 4839902 : cstate->raw_fields[fieldno] = output_ptr;
4379 :
4380 : /*
4381 : * Scan data for field.
4382 : *
4383 : * Note that in this loop, we are scanning to locate the end of field
4384 : * and also speculatively performing de-escaping. Once we find the
4385 : * end-of-field, we can match the raw field contents against the null
4386 : * marker string. Only after that comparison fails do we know that
4387 : * de-escaping is actually the right thing to do; therefore we *must
4388 : * not* throw any syntax errors before we've done the null-marker
4389 : * check.
4390 : */
4391 : for (;;)
4392 24932858 : {
4393 : char c;
4394 :
4395 29772760 : end_ptr = cur_ptr;
4396 29772760 : if (cur_ptr >= line_end_ptr)
4397 1155786 : break;
4398 28616974 : c = *cur_ptr++;
4399 28616974 : if (c == delimc)
4400 : {
4401 3684116 : found_delim = true;
4402 3684116 : break;
4403 : }
4404 24932858 : if (c == '\\')
4405 : {
4406 6170 : if (cur_ptr >= line_end_ptr)
4407 0 : break;
4408 6170 : c = *cur_ptr++;
4409 6170 : switch (c)
4410 : {
4411 8 : case '0':
4412 : case '1':
4413 : case '2':
4414 : case '3':
4415 : case '4':
4416 : case '5':
4417 : case '6':
4418 : case '7':
4419 0 : {
4420 : /* handle \013 */
4421 : int val;
4422 :
4423 8 : val = OCTVALUE(c);
4424 8 : if (cur_ptr < line_end_ptr)
4425 : {
4426 4 : c = *cur_ptr;
4427 4 : if (ISOCTAL(c))
4428 : {
4429 0 : cur_ptr++;
4430 0 : val = (val << 3) + OCTVALUE(c);
4431 0 : if (cur_ptr < line_end_ptr)
4432 : {
4433 0 : c = *cur_ptr;
4434 0 : if (ISOCTAL(c))
4435 : {
4436 0 : cur_ptr++;
4437 0 : val = (val << 3) + OCTVALUE(c);
4438 : }
4439 : }
4440 : }
4441 : }
4442 8 : c = val & 0377;
4443 8 : if (c == '\0' || IS_HIGHBIT_SET(c))
4444 8 : saw_non_ascii = true;
4445 : }
4446 8 : break;
4447 8 : case 'x':
4448 : /* Handle \x3F */
4449 8 : if (cur_ptr < line_end_ptr)
4450 : {
4451 4 : char hexchar = *cur_ptr;
4452 :
4453 4 : if (isxdigit((unsigned char) hexchar))
4454 : {
4455 0 : int val = GetDecimalFromHex(hexchar);
4456 :
4457 0 : cur_ptr++;
4458 0 : if (cur_ptr < line_end_ptr)
4459 : {
4460 0 : hexchar = *cur_ptr;
4461 0 : if (isxdigit((unsigned char) hexchar))
4462 : {
4463 0 : cur_ptr++;
4464 0 : val = (val << 4) + GetDecimalFromHex(hexchar);
4465 : }
4466 : }
4467 0 : c = val & 0xff;
4468 0 : if (c == '\0' || IS_HIGHBIT_SET(c))
4469 0 : saw_non_ascii = true;
4470 : }
4471 : }
4472 8 : break;
4473 0 : case 'b':
4474 0 : c = '\b';
4475 0 : break;
4476 0 : case 'f':
4477 0 : c = '\f';
4478 0 : break;
4479 2034 : case 'n':
4480 2034 : c = '\n';
4481 2034 : break;
4482 0 : case 'r':
4483 0 : c = '\r';
4484 0 : break;
4485 0 : case 't':
4486 0 : c = '\t';
4487 0 : break;
4488 0 : case 'v':
4489 0 : c = '\v';
4490 0 : break;
4491 :
4492 : /*
4493 : * in all other cases, take the char after '\'
4494 : * literally
4495 : */
4496 : }
4497 24926688 : }
4498 :
4499 : /* Add c to output string */
4500 24932858 : *output_ptr++ = c;
4501 : }
4502 :
4503 : /* Check whether raw input matched null marker */
4504 4839902 : input_len = end_ptr - start_ptr;
4505 4839902 : if (input_len == cstate->null_print_len &&
4506 341962 : strncmp(start_ptr, cstate->null_print, input_len) == 0)
4507 4096 : cstate->raw_fields[fieldno] = NULL;
4508 : else
4509 : {
4510 : /*
4511 : * At this point we know the field is supposed to contain data.
4512 : *
4513 : * If we de-escaped any non-7-bit-ASCII chars, make sure the
4514 : * resulting string is valid data for the db encoding.
4515 : */
4516 4835806 : if (saw_non_ascii)
4517 : {
4518 0 : char *fld = cstate->raw_fields[fieldno];
4519 :
4520 0 : pg_verifymbstr(fld, output_ptr - fld, false);
4521 : }
4522 : }
4523 :
4524 : /* Terminate attribute value in output area */
4525 4839902 : *output_ptr++ = '\0';
4526 :
4527 4839902 : fieldno++;
4528 : /* Done if we hit EOL instead of a delim */
4529 4839902 : if (!found_delim)
4530 1155786 : break;
4531 : }
4532 :
4533 : /* Clean up state of attribute_buf */
4534 1155786 : output_ptr--;
4535 : Assert(*output_ptr == '\0');
4536 1155786 : cstate->attribute_buf.len = (output_ptr - cstate->attribute_buf.data);
4537 :
4538 1155786 : return fieldno;
4539 : }
4540 :
4541 : /*
4542 : * Parse the current line into separate attributes (fields),
4543 : * performing de-escaping as needed. This has exactly the same API as
4544 : * CopyReadAttributesText, except we parse the fields according to
4545 : * "standard" (i.e. common) CSV usage.
4546 : */
4547 : static int
4548 238 : CopyReadAttributesCSV(CopyState cstate)
4549 : {
4550 238 : char delimc = cstate->delim[0];
4551 238 : char quotec = cstate->quote[0];
4552 238 : char escapec = cstate->escape[0];
4553 : int fieldno;
4554 : char *output_ptr;
4555 : char *cur_ptr;
4556 : char *line_end_ptr;
4557 :
4558 : /*
4559 : * We need a special case for zero-column tables: check that the input
4560 : * line is empty, and return.
4561 : */
4562 238 : if (cstate->max_fields <= 0)
4563 : {
4564 0 : if (cstate->line_buf.len != 0)
4565 0 : ereport(ERROR,
4566 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4567 : errmsg("extra data after last expected column")));
4568 0 : return 0;
4569 : }
4570 :
4571 238 : resetStringInfo(&cstate->attribute_buf);
4572 :
4573 : /*
4574 : * The de-escaped attributes will certainly not be longer than the input
4575 : * data line, so we can just force attribute_buf to be large enough and
4576 : * then transfer data without any checks for enough space. We need to do
4577 : * it this way because enlarging attribute_buf mid-stream would invalidate
4578 : * pointers already stored into cstate->raw_fields[].
4579 : */
4580 238 : if (cstate->attribute_buf.maxlen <= cstate->line_buf.len)
4581 0 : enlargeStringInfo(&cstate->attribute_buf, cstate->line_buf.len);
4582 238 : output_ptr = cstate->attribute_buf.data;
4583 :
4584 : /* set pointer variables for loop */
4585 238 : cur_ptr = cstate->line_buf.data;
4586 238 : line_end_ptr = cstate->line_buf.data + cstate->line_buf.len;
4587 :
4588 : /* Outer loop iterates over fields */
4589 238 : fieldno = 0;
4590 : for (;;)
4591 246 : {
4592 484 : bool found_delim = false;
4593 484 : bool saw_quote = false;
4594 : char *start_ptr;
4595 : char *end_ptr;
4596 : int input_len;
4597 :
4598 : /* Make sure there is enough space for the next value */
4599 484 : if (fieldno >= cstate->max_fields)
4600 : {
4601 0 : cstate->max_fields *= 2;
4602 0 : cstate->raw_fields =
4603 0 : repalloc(cstate->raw_fields, cstate->max_fields * sizeof(char *));
4604 : }
4605 :
4606 : /* Remember start of field on both input and output sides */
4607 484 : start_ptr = cur_ptr;
4608 484 : cstate->raw_fields[fieldno] = output_ptr;
4609 :
4610 : /*
4611 : * Scan data for field,
4612 : *
4613 : * The loop starts in "not quote" mode and then toggles between that
4614 : * and "in quote" mode. The loop exits normally if it is in "not
4615 : * quote" mode and a delimiter or line end is seen.
4616 : */
4617 : for (;;)
4618 126 : {
4619 : char c;
4620 :
4621 : /* Not in quote */
4622 : for (;;)
4623 : {
4624 2050 : end_ptr = cur_ptr;
4625 1330 : if (cur_ptr >= line_end_ptr)
4626 238 : goto endfield;
4627 1092 : c = *cur_ptr++;
4628 : /* unquoted field delimiter */
4629 1092 : if (c == delimc)
4630 : {
4631 246 : found_delim = true;
4632 246 : goto endfield;
4633 : }
4634 : /* start of quoted field (or part of field) */
4635 846 : if (c == quotec)
4636 : {
4637 126 : saw_quote = true;
4638 126 : break;
4639 : }
4640 : /* Add c to output string */
4641 720 : *output_ptr++ = c;
4642 : }
4643 :
4644 : /* In quote */
4645 : for (;;)
4646 : {
4647 782 : end_ptr = cur_ptr;
4648 908 : if (cur_ptr >= line_end_ptr)
4649 0 : ereport(ERROR,
4650 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4651 : errmsg("unterminated CSV quoted field")));
4652 :
4653 908 : c = *cur_ptr++;
4654 :
4655 : /* escape within a quoted field */
4656 908 : if (c == escapec)
4657 : {
4658 : /*
4659 : * peek at the next char if available, and escape it if it
4660 : * is an escape char or a quote char
4661 : */
4662 62 : if (cur_ptr < line_end_ptr)
4663 : {
4664 42 : char nextc = *cur_ptr;
4665 :
4666 42 : if (nextc == escapec || nextc == quotec)
4667 : {
4668 16 : *output_ptr++ = nextc;
4669 16 : cur_ptr++;
4670 16 : continue;
4671 : }
4672 : }
4673 : }
4674 :
4675 : /*
4676 : * end of quoted field. Must do this test after testing for
4677 : * escape in case quote char and escape char are the same
4678 : * (which is the common case).
4679 : */
4680 892 : if (c == quotec)
4681 126 : break;
4682 :
4683 : /* Add c to output string */
4684 766 : *output_ptr++ = c;
4685 : }
4686 : }
4687 484 : endfield:
4688 :
4689 : /* Terminate attribute value in output area */
4690 484 : *output_ptr++ = '\0';
4691 :
4692 : /* Check whether raw input matched null marker */
4693 484 : input_len = end_ptr - start_ptr;
4694 484 : if (!saw_quote && input_len == cstate->null_print_len &&
4695 26 : strncmp(start_ptr, cstate->null_print, input_len) == 0)
4696 26 : cstate->raw_fields[fieldno] = NULL;
4697 :
4698 484 : fieldno++;
4699 : /* Done if we hit EOL instead of a delim */
4700 484 : if (!found_delim)
4701 238 : break;
4702 : }
4703 :
4704 : /* Clean up state of attribute_buf */
4705 238 : output_ptr--;
4706 : Assert(*output_ptr == '\0');
4707 238 : cstate->attribute_buf.len = (output_ptr - cstate->attribute_buf.data);
4708 :
4709 238 : return fieldno;
4710 : }
4711 :
4712 :
4713 : /*
4714 : * Read a binary attribute
4715 : */
4716 : static Datum
4717 84 : CopyReadBinaryAttribute(CopyState cstate,
4718 : int column_no, FmgrInfo *flinfo,
4719 : Oid typioparam, int32 typmod,
4720 : bool *isnull)
4721 : {
4722 : int32 fld_size;
4723 : Datum result;
4724 :
4725 84 : if (!CopyGetInt32(cstate, &fld_size))
4726 0 : ereport(ERROR,
4727 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4728 : errmsg("unexpected EOF in COPY data")));
4729 84 : if (fld_size == -1)
4730 : {
4731 20 : *isnull = true;
4732 20 : return ReceiveFunctionCall(flinfo, NULL, typioparam, typmod);
4733 : }
4734 64 : if (fld_size < 0)
4735 0 : ereport(ERROR,
4736 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4737 : errmsg("invalid field size")));
4738 :
4739 : /* reset attribute_buf to empty, and load raw data in it */
4740 64 : resetStringInfo(&cstate->attribute_buf);
4741 :
4742 64 : enlargeStringInfo(&cstate->attribute_buf, fld_size);
4743 64 : if (CopyGetData(cstate, cstate->attribute_buf.data,
4744 64 : fld_size, fld_size) != fld_size)
4745 0 : ereport(ERROR,
4746 : (errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
4747 : errmsg("unexpected EOF in COPY data")));
4748 :
4749 64 : cstate->attribute_buf.len = fld_size;
4750 64 : cstate->attribute_buf.data[fld_size] = '\0';
4751 :
4752 : /* Call the column type's binary input converter */
4753 64 : result = ReceiveFunctionCall(flinfo, &cstate->attribute_buf,
4754 : typioparam, typmod);
4755 :
4756 : /* Trouble if it didn't eat the whole buffer */
4757 64 : if (cstate->attribute_buf.cursor != cstate->attribute_buf.len)
4758 0 : ereport(ERROR,
4759 : (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
4760 : errmsg("incorrect binary data format")));
4761 :
4762 64 : *isnull = false;
4763 64 : return result;
4764 : }
4765 :
4766 : /*
4767 : * Send text representation of one attribute, with conversion and escaping
4768 : */
4769 : #define DUMPSOFAR() \
4770 : do { \
4771 : if (ptr > start) \
4772 : CopySendData(cstate, start, ptr - start); \
4773 : } while (0)
4774 :
4775 : static void
4776 4560556 : CopyAttributeOutText(CopyState cstate, char *string)
4777 : {
4778 : char *ptr;
4779 : char *start;
4780 : char c;
4781 4560556 : char delimc = cstate->delim[0];
4782 :
4783 4560556 : if (cstate->need_transcoding)
4784 4560532 : ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding);
4785 : else
4786 24 : ptr = string;
4787 :
4788 : /*
4789 : * We have to grovel through the string searching for control characters
4790 : * and instances of the delimiter character. In most cases, though, these
4791 : * are infrequent. To avoid overhead from calling CopySendData once per
4792 : * character, we dump out all characters between escaped characters in a
4793 : * single call. The loop invariant is that the data from "start" to "ptr"
4794 : * can be sent literally, but hasn't yet been.
4795 : *
4796 : * We can skip pg_encoding_mblen() overhead when encoding is safe, because
4797 : * in valid backend encodings, extra bytes of a multibyte character never
4798 : * look like ASCII. This loop is sufficiently performance-critical that
4799 : * it's worth making two copies of it to get the IS_HIGHBIT_SET() test out
4800 : * of the normal safe-encoding path.
4801 : */
4802 4560556 : if (cstate->encoding_embeds_ascii)
4803 : {
4804 0 : start = ptr;
4805 0 : while ((c = *ptr) != '\0')
4806 : {
4807 0 : if ((unsigned char) c < (unsigned char) 0x20)
4808 : {
4809 : /*
4810 : * \r and \n must be escaped, the others are traditional. We
4811 : * prefer to dump these using the C-like notation, rather than
4812 : * a backslash and the literal character, because it makes the
4813 : * dump file a bit more proof against Microsoftish data
4814 : * mangling.
4815 : */
4816 0 : switch (c)
4817 : {
4818 0 : case '\b':
4819 0 : c = 'b';
4820 0 : break;
4821 0 : case '\f':
4822 0 : c = 'f';
4823 0 : break;
4824 0 : case '\n':
4825 0 : c = 'n';
4826 0 : break;
4827 0 : case '\r':
4828 0 : c = 'r';
4829 0 : break;
4830 0 : case '\t':
4831 0 : c = 't';
4832 0 : break;
4833 0 : case '\v':
4834 0 : c = 'v';
4835 0 : break;
4836 0 : default:
4837 : /* If it's the delimiter, must backslash it */
4838 0 : if (c == delimc)
4839 0 : break;
4840 : /* All ASCII control chars are length 1 */
4841 0 : ptr++;
4842 0 : continue; /* fall to end of loop */
4843 : }
4844 : /* if we get here, we need to convert the control char */
4845 0 : DUMPSOFAR();
4846 0 : CopySendChar(cstate, '\\');
4847 0 : CopySendChar(cstate, c);
4848 0 : start = ++ptr; /* do not include char in next run */
4849 : }
4850 0 : else if (c == '\\' || c == delimc)
4851 : {
4852 0 : DUMPSOFAR();
4853 0 : CopySendChar(cstate, '\\');
4854 0 : start = ptr++; /* we include char in next run */
4855 : }
4856 0 : else if (IS_HIGHBIT_SET(c))
4857 0 : ptr += pg_encoding_mblen(cstate->file_encoding, ptr);
4858 : else
4859 0 : ptr++;
4860 : }
4861 : }
4862 : else
4863 : {
4864 4560556 : start = ptr;
4865 47084862 : while ((c = *ptr) != '\0')
4866 : {
4867 42524306 : if ((unsigned char) c < (unsigned char) 0x20)
4868 : {
4869 : /*
4870 : * \r and \n must be escaped, the others are traditional. We
4871 : * prefer to dump these using the C-like notation, rather than
4872 : * a backslash and the literal character, because it makes the
4873 : * dump file a bit more proof against Microsoftish data
4874 : * mangling.
4875 : */
4876 2264 : switch (c)
4877 : {
4878 0 : case '\b':
4879 0 : c = 'b';
4880 0 : break;
4881 0 : case '\f':
4882 0 : c = 'f';
4883 0 : break;
4884 2264 : case '\n':
4885 2264 : c = 'n';
4886 2264 : break;
4887 0 : case '\r':
4888 0 : c = 'r';
4889 0 : break;
4890 0 : case '\t':
4891 0 : c = 't';
4892 0 : break;
4893 0 : case '\v':
4894 0 : c = 'v';
4895 0 : break;
4896 0 : default:
4897 : /* If it's the delimiter, must backslash it */
4898 0 : if (c == delimc)
4899 0 : break;
4900 : /* All ASCII control chars are length 1 */
4901 0 : ptr++;
4902 0 : continue; /* fall to end of loop */
4903 : }
4904 : /* if we get here, we need to convert the control char */
4905 2264 : DUMPSOFAR();
4906 2264 : CopySendChar(cstate, '\\');
4907 2264 : CopySendChar(cstate, c);
4908 2264 : start = ++ptr; /* do not include char in next run */
4909 : }
4910 42522042 : else if (c == '\\' || c == delimc)
4911 : {
4912 456 : DUMPSOFAR();
4913 456 : CopySendChar(cstate, '\\');
4914 456 : start = ptr++; /* we include char in next run */
4915 : }
4916 : else
4917 42521586 : ptr++;
4918 : }
4919 : }
4920 :
4921 4560556 : DUMPSOFAR();
4922 4560556 : }
4923 :
4924 : /*
4925 : * Send text representation of one attribute, with conversion and
4926 : * CSV-style escaping
4927 : */
4928 : static void
4929 396 : CopyAttributeOutCSV(CopyState cstate, char *string,
4930 : bool use_quote, bool single_attr)
4931 : {
4932 : char *ptr;
4933 : char *start;
4934 : char c;
4935 396 : char delimc = cstate->delim[0];
4936 396 : char quotec = cstate->quote[0];
4937 396 : char escapec = cstate->escape[0];
4938 :
4939 : /* force quoting if it matches null_print (before conversion!) */
4940 396 : if (!use_quote && strcmp(string, cstate->null_print) == 0)
4941 36 : use_quote = true;
4942 :
4943 396 : if (cstate->need_transcoding)
4944 396 : ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding);
4945 : else
4946 0 : ptr = string;
4947 :
4948 : /*
4949 : * Make a preliminary pass to discover if it needs quoting
4950 : */
4951 396 : if (!use_quote)
4952 : {
4953 : /*
4954 : * Because '\.' can be a data value, quote it if it appears alone on a
4955 : * line so it is not interpreted as the end-of-data marker.
4956 : */
4957 272 : if (single_attr && strcmp(ptr, "\\.") == 0)
4958 4 : use_quote = true;
4959 : else
4960 : {
4961 268 : char *tptr = ptr;
4962 :
4963 1408 : while ((c = *tptr) != '\0')
4964 : {
4965 1228 : if (c == delimc || c == quotec || c == '\n' || c == '\r')
4966 : {
4967 88 : use_quote = true;
4968 88 : break;
4969 : }
4970 1140 : if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii)
4971 0 : tptr += pg_encoding_mblen(cstate->file_encoding, tptr);
4972 : else
4973 1140 : tptr++;
4974 : }
4975 : }
4976 : }
4977 :
4978 396 : if (use_quote)
4979 : {
4980 216 : CopySendChar(cstate, quotec);
4981 :
4982 : /*
4983 : * We adopt the same optimization strategy as in CopyAttributeOutText
4984 : */
4985 216 : start = ptr;
4986 1692 : while ((c = *ptr) != '\0')
4987 : {
4988 1476 : if (c == quotec || c == escapec)
4989 : {
4990 104 : DUMPSOFAR();
4991 104 : CopySendChar(cstate, escapec);
4992 104 : start = ptr; /* we include char in next run */
4993 : }
4994 1476 : if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii)
4995 0 : ptr += pg_encoding_mblen(cstate->file_encoding, ptr);
4996 : else
4997 1476 : ptr++;
4998 : }
4999 216 : DUMPSOFAR();
5000 :
5001 216 : CopySendChar(cstate, quotec);
5002 : }
5003 : else
5004 : {
5005 : /* If it doesn't need quoting, we can just dump it as-is */
5006 180 : CopySendString(cstate, ptr);
5007 : }
5008 396 : }
5009 :
5010 : /*
5011 : * CopyGetAttnums - build an integer list of attnums to be copied
5012 : *
5013 : * The input attnamelist is either the user-specified column list,
5014 : * or NIL if there was none (in which case we want all the non-dropped
5015 : * columns).
5016 : *
5017 : * We don't include generated columns in the generated full list and we don't
5018 : * allow them to be specified explicitly. They don't make sense for COPY
5019 : * FROM, but we could possibly allow them for COPY TO. But this way it's at
5020 : * least ensured that whatever we copy out can be copied back in.
5021 : *
5022 : * rel can be NULL ... it's only used for error reports.
5023 : */
5024 : static List *
5025 8080 : CopyGetAttnums(TupleDesc tupDesc, Relation rel, List *attnamelist)
5026 : {
5027 8080 : List *attnums = NIL;
5028 :
5029 8080 : if (attnamelist == NIL)
5030 : {
5031 : /* Generate default column list */
5032 1966 : int attr_count = tupDesc->natts;
5033 : int i;
5034 :
5035 6556 : for (i = 0; i < attr_count; i++)
5036 : {
5037 4590 : if (TupleDescAttr(tupDesc, i)->attisdropped)
5038 48 : continue;
5039 4542 : if (TupleDescAttr(tupDesc, i)->attgenerated)
5040 38 : continue;
5041 4504 : attnums = lappend_int(attnums, i + 1);
5042 : }
5043 : }
5044 : else
5045 : {
5046 : /* Validate the user-supplied list and extract attnums */
5047 : ListCell *l;
5048 :
5049 31884 : foreach(l, attnamelist)
5050 : {
5051 25810 : char *name = strVal(lfirst(l));
5052 : int attnum;
5053 : int i;
5054 :
5055 : /* Lookup column name */
5056 25810 : attnum = InvalidAttrNumber;
5057 4910442 : for (i = 0; i < tupDesc->natts; i++)
5058 : {
5059 4910422 : Form_pg_attribute att = TupleDescAttr(tupDesc, i);
5060 :
5061 4910422 : if (att->attisdropped)
5062 288 : continue;
5063 4910134 : if (namestrcmp(&(att->attname), name) == 0)
5064 : {
5065 25790 : if (att->attgenerated)
5066 16 : ereport(ERROR,
5067 : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
5068 : errmsg("column \"%s\" is a generated column",
5069 : name),
5070 : errdetail("Generated columns cannot be used in COPY.")));
5071 25774 : attnum = att->attnum;
5072 25774 : break;
5073 : }
5074 : }
5075 25794 : if (attnum == InvalidAttrNumber)
5076 : {
5077 20 : if (rel != NULL)
5078 20 : ereport(ERROR,
5079 : (errcode(ERRCODE_UNDEFINED_COLUMN),
5080 : errmsg("column \"%s\" of relation \"%s\" does not exist",
5081 : name, RelationGetRelationName(rel))));
5082 : else
5083 0 : ereport(ERROR,
5084 : (errcode(ERRCODE_UNDEFINED_COLUMN),
5085 : errmsg("column \"%s\" does not exist",
5086 : name)));
5087 : }
5088 : /* Check for duplicates */
5089 25774 : if (list_member_int(attnums, attnum))
5090 4 : ereport(ERROR,
5091 : (errcode(ERRCODE_DUPLICATE_COLUMN),
5092 : errmsg("column \"%s\" specified more than once",
5093 : name)));
5094 25770 : attnums = lappend_int(attnums, attnum);
5095 : }
5096 : }
5097 :
5098 8040 : return attnums;
5099 : }
5100 :
5101 :
5102 : /*
5103 : * copy_dest_startup --- executor startup
5104 : */
5105 : static void
5106 140 : copy_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
5107 : {
5108 : /* no-op */
5109 140 : }
5110 :
5111 : /*
5112 : * copy_dest_receive --- receive one tuple
5113 : */
5114 : static bool
5115 4468 : copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
5116 : {
5117 4468 : DR_copy *myState = (DR_copy *) self;
5118 4468 : CopyState cstate = myState->cstate;
5119 :
5120 : /* Send the data */
5121 4468 : CopyOneRowTo(cstate, slot);
5122 4468 : myState->processed++;
5123 :
5124 4468 : return true;
5125 : }
5126 :
5127 : /*
5128 : * copy_dest_shutdown --- executor end
5129 : */
5130 : static void
5131 140 : copy_dest_shutdown(DestReceiver *self)
5132 : {
5133 : /* no-op */
5134 140 : }
5135 :
5136 : /*
5137 : * copy_dest_destroy --- release DestReceiver object
5138 : */
5139 : static void
5140 0 : copy_dest_destroy(DestReceiver *self)
5141 : {
5142 0 : pfree(self);
5143 0 : }
5144 :
5145 : /*
5146 : * CreateCopyDestReceiver -- create a suitable DestReceiver object
5147 : */
5148 : DestReceiver *
5149 144 : CreateCopyDestReceiver(void)
5150 : {
5151 144 : DR_copy *self = (DR_copy *) palloc(sizeof(DR_copy));
5152 :
5153 144 : self->pub.receiveSlot = copy_dest_receive;
5154 144 : self->pub.rStartup = copy_dest_startup;
5155 144 : self->pub.rShutdown = copy_dest_shutdown;
5156 144 : self->pub.rDestroy = copy_dest_destroy;
5157 144 : self->pub.mydest = DestCopyOut;
5158 :
5159 144 : self->cstate = NULL; /* will be set later */
5160 144 : self->processed = 0;
5161 :
5162 144 : return (DestReceiver *) self;
5163 : }
|
