Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * copyto.c
4 : : * COPY <table> TO file/program/client
5 : : *
6 : : * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : *
10 : : * IDENTIFICATION
11 : : * src/backend/commands/copyto.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/tableam.h"
22 : : #include "commands/copyapi.h"
23 : : #include "commands/progress.h"
24 : : #include "executor/execdesc.h"
25 : : #include "executor/executor.h"
26 : : #include "executor/tuptable.h"
27 : : #include "libpq/libpq.h"
28 : : #include "libpq/pqformat.h"
29 : : #include "mb/pg_wchar.h"
30 : : #include "miscadmin.h"
31 : : #include "pgstat.h"
32 : : #include "storage/fd.h"
33 : : #include "tcop/tcopprot.h"
34 : : #include "utils/lsyscache.h"
35 : : #include "utils/memutils.h"
36 : : #include "utils/rel.h"
37 : : #include "utils/snapmgr.h"
38 : :
39 : : /*
40 : : * Represents the different dest cases we need to worry about at
41 : : * the bottom level
42 : : */
43 : : typedef enum CopyDest
44 : : {
45 : : COPY_FILE, /* to file (or a piped program) */
46 : : COPY_FRONTEND, /* to frontend */
47 : : COPY_CALLBACK, /* to callback function */
48 : : } CopyDest;
49 : :
50 : : /*
51 : : * This struct contains all the state variables used throughout a COPY TO
52 : : * operation.
53 : : *
54 : : * Multi-byte encodings: all supported client-side encodings encode multi-byte
55 : : * characters by having the first byte's high bit set. Subsequent bytes of the
56 : : * character can have the high bit not set. When scanning data in such an
57 : : * encoding to look for a match to a single-byte (ie ASCII) character, we must
58 : : * use the full pg_encoding_mblen() machinery to skip over multibyte
59 : : * characters, else we might find a false match to a trailing byte. In
60 : : * supported server encodings, there is no possibility of a false match, and
61 : : * it's faster to make useless comparisons to trailing bytes than it is to
62 : : * invoke pg_encoding_mblen() to skip over them. encoding_embeds_ascii is true
63 : : * when we have to do it the hard way.
64 : : */
65 : : typedef struct CopyToStateData
66 : : {
67 : : /* format-specific routines */
68 : : const CopyToRoutine *routine;
69 : :
70 : : /* low-level state data */
71 : : CopyDest copy_dest; /* type of copy source/destination */
72 : : FILE *copy_file; /* used if copy_dest == COPY_FILE */
73 : : StringInfo fe_msgbuf; /* used for all dests during COPY TO */
74 : :
75 : : int file_encoding; /* file or remote side's character encoding */
76 : : bool need_transcoding; /* file encoding diff from server? */
77 : : bool encoding_embeds_ascii; /* ASCII can be non-first byte? */
78 : :
79 : : /* parameters from the COPY command */
80 : : Relation rel; /* relation to copy to */
81 : : QueryDesc *queryDesc; /* executable query to copy from */
82 : : List *attnumlist; /* integer list of attnums to copy */
83 : : char *filename; /* filename, or NULL for STDOUT */
84 : : bool is_program; /* is 'filename' a program to popen? */
85 : : copy_data_dest_cb data_dest_cb; /* function for writing data */
86 : :
87 : : CopyFormatOptions opts;
88 : : Node *whereClause; /* WHERE condition (or NULL) */
89 : :
90 : : /*
91 : : * Working state
92 : : */
93 : : MemoryContext copycontext; /* per-copy execution context */
94 : :
95 : : FmgrInfo *out_functions; /* lookup info for output functions */
96 : : MemoryContext rowcontext; /* per-row evaluation context */
97 : : uint64 bytes_processed; /* number of bytes processed so far */
98 : : } CopyToStateData;
99 : :
100 : : /* DestReceiver for COPY (query) TO */
101 : : typedef struct
102 : : {
103 : : DestReceiver pub; /* publicly-known function pointers */
104 : : CopyToState cstate; /* CopyToStateData for the command */
105 : : uint64 processed; /* # of tuples processed */
106 : : } DR_copy;
107 : :
108 : : /* NOTE: there's a copy of this in copyfromparse.c */
109 : : static const char BinarySignature[11] = "PGCOPY\n\377\r\n\0";
110 : :
111 : :
112 : : /* non-export function prototypes */
113 : : static void EndCopy(CopyToState cstate);
114 : : static void ClosePipeToProgram(CopyToState cstate);
115 : : static void CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot);
116 : : static void CopyAttributeOutText(CopyToState cstate, const char *string);
117 : : static void CopyAttributeOutCSV(CopyToState cstate, const char *string,
118 : : bool use_quote);
119 : :
120 : : /* built-in format-specific routines */
121 : : static void CopyToTextLikeStart(CopyToState cstate, TupleDesc tupDesc);
122 : : static void CopyToTextLikeOutFunc(CopyToState cstate, Oid atttypid, FmgrInfo *finfo);
123 : : static void CopyToTextOneRow(CopyToState cstate, TupleTableSlot *slot);
124 : : static void CopyToCSVOneRow(CopyToState cstate, TupleTableSlot *slot);
125 : : static void CopyToTextLikeOneRow(CopyToState cstate, TupleTableSlot *slot,
126 : : bool is_csv);
127 : : static void CopyToTextLikeEnd(CopyToState cstate);
128 : : static void CopyToBinaryStart(CopyToState cstate, TupleDesc tupDesc);
129 : : static void CopyToBinaryOutFunc(CopyToState cstate, Oid atttypid, FmgrInfo *finfo);
130 : : static void CopyToBinaryOneRow(CopyToState cstate, TupleTableSlot *slot);
131 : : static void CopyToBinaryEnd(CopyToState cstate);
132 : :
133 : : /* Low-level communications functions */
134 : : static void SendCopyBegin(CopyToState cstate);
135 : : static void SendCopyEnd(CopyToState cstate);
136 : : static void CopySendData(CopyToState cstate, const void *databuf, int datasize);
137 : : static void CopySendString(CopyToState cstate, const char *str);
138 : : static void CopySendChar(CopyToState cstate, char c);
139 : : static void CopySendEndOfRow(CopyToState cstate);
140 : : static void CopySendTextLikeEndOfRow(CopyToState cstate);
141 : : static void CopySendInt32(CopyToState cstate, int32 val);
142 : : static void CopySendInt16(CopyToState cstate, int16 val);
143 : :
144 : : /*
145 : : * COPY TO routines for built-in formats.
146 : : *
147 : : * CSV and text formats share the same TextLike routines except for the
148 : : * one-row callback.
149 : : */
150 : :
151 : : /* text format */
152 : : static const CopyToRoutine CopyToRoutineText = {
153 : : .CopyToStart = CopyToTextLikeStart,
154 : : .CopyToOutFunc = CopyToTextLikeOutFunc,
155 : : .CopyToOneRow = CopyToTextOneRow,
156 : : .CopyToEnd = CopyToTextLikeEnd,
157 : : };
158 : :
159 : : /* CSV format */
160 : : static const CopyToRoutine CopyToRoutineCSV = {
161 : : .CopyToStart = CopyToTextLikeStart,
162 : : .CopyToOutFunc = CopyToTextLikeOutFunc,
163 : : .CopyToOneRow = CopyToCSVOneRow,
164 : : .CopyToEnd = CopyToTextLikeEnd,
165 : : };
166 : :
167 : : /* binary format */
168 : : static const CopyToRoutine CopyToRoutineBinary = {
169 : : .CopyToStart = CopyToBinaryStart,
170 : : .CopyToOutFunc = CopyToBinaryOutFunc,
171 : : .CopyToOneRow = CopyToBinaryOneRow,
172 : : .CopyToEnd = CopyToBinaryEnd,
173 : : };
174 : :
175 : : /* Return a COPY TO routine for the given options */
176 : : static const CopyToRoutine *
186 msawada@postgresql.o 177 :CBC 4891 : CopyToGetRoutine(const CopyFormatOptions *opts)
178 : : {
179 [ + + ]: 4891 : if (opts->csv_mode)
191 180 : 63 : return &CopyToRoutineCSV;
186 181 [ + + ]: 4828 : else if (opts->binary)
191 182 : 8 : return &CopyToRoutineBinary;
183 : :
184 : : /* default is text */
185 : 4820 : return &CopyToRoutineText;
186 : : }
187 : :
188 : : /* Implementation of the start callback for text and CSV formats */
189 : : static void
190 : 4816 : CopyToTextLikeStart(CopyToState cstate, TupleDesc tupDesc)
191 : : {
192 : : /*
193 : : * For non-binary copy, we need to convert null_print to file encoding,
194 : : * because it will be sent directly with CopySendString.
195 : : */
196 [ + + ]: 4816 : if (cstate->need_transcoding)
197 : 1 : cstate->opts.null_print_client = pg_server_to_any(cstate->opts.null_print,
198 : : cstate->opts.null_print_len,
199 : : cstate->file_encoding);
200 : :
201 : : /* if a header has been requested send the line */
65 fujii@postgresql.org 202 [ + + ]:GNC 4816 : if (cstate->opts.header_line == COPY_HEADER_TRUE)
203 : : {
204 : : ListCell *cur;
191 msawada@postgresql.o 205 :CBC 15 : bool hdr_delim = false;
206 : :
207 [ + - + + : 39 : foreach(cur, cstate->attnumlist)
+ + ]
208 : : {
209 : 24 : int attnum = lfirst_int(cur);
210 : : char *colname;
211 : :
212 [ + + ]: 24 : if (hdr_delim)
213 : 9 : CopySendChar(cstate, cstate->opts.delim[0]);
214 : 24 : hdr_delim = true;
215 : :
216 : 24 : colname = NameStr(TupleDescAttr(tupDesc, attnum - 1)->attname);
217 : :
218 [ + + ]: 24 : if (cstate->opts.csv_mode)
219 : 12 : CopyAttributeOutCSV(cstate, colname, false);
220 : : else
221 : 12 : CopyAttributeOutText(cstate, colname);
222 : : }
223 : :
224 : 15 : CopySendTextLikeEndOfRow(cstate);
225 : : }
226 : 4816 : }
227 : :
228 : : /*
229 : : * Implementation of the outfunc callback for text and CSV formats. Assign
230 : : * the output function data to the given *finfo.
231 : : */
232 : : static void
233 : 16316 : CopyToTextLikeOutFunc(CopyToState cstate, Oid atttypid, FmgrInfo *finfo)
234 : : {
235 : : Oid func_oid;
236 : : bool is_varlena;
237 : :
238 : : /* Set output function for an attribute */
239 : 16316 : getTypeOutputInfo(atttypid, &func_oid, &is_varlena);
240 : 16316 : fmgr_info(func_oid, finfo);
241 : 16316 : }
242 : :
243 : : /* Implementation of the per-row callback for text format */
244 : : static void
245 : 1829591 : CopyToTextOneRow(CopyToState cstate, TupleTableSlot *slot)
246 : : {
247 : 1829591 : CopyToTextLikeOneRow(cstate, slot, false);
248 : 1829591 : }
249 : :
250 : : /* Implementation of the per-row callback for CSV format */
251 : : static void
252 : 165 : CopyToCSVOneRow(CopyToState cstate, TupleTableSlot *slot)
253 : : {
254 : 165 : CopyToTextLikeOneRow(cstate, slot, true);
255 : 165 : }
256 : :
257 : : /*
258 : : * Workhorse for CopyToTextOneRow() and CopyToCSVOneRow().
259 : : *
260 : : * We use pg_attribute_always_inline to reduce function call overhead
261 : : * and to help compilers to optimize away the 'is_csv' condition.
262 : : */
263 : : static pg_attribute_always_inline void
264 : 1829756 : CopyToTextLikeOneRow(CopyToState cstate,
265 : : TupleTableSlot *slot,
266 : : bool is_csv)
267 : : {
268 : 1829756 : bool need_delim = false;
269 : 1829756 : FmgrInfo *out_functions = cstate->out_functions;
270 : :
271 [ + + + + : 10747599 : foreach_int(attnum, cstate->attnumlist)
+ + ]
272 : : {
273 : 7088087 : Datum value = slot->tts_values[attnum - 1];
274 : 7088087 : bool isnull = slot->tts_isnull[attnum - 1];
275 : :
276 [ + + ]: 7088087 : if (need_delim)
277 : 5258408 : CopySendChar(cstate, cstate->opts.delim[0]);
278 : 7088087 : need_delim = true;
279 : :
280 [ + + ]: 7088087 : if (isnull)
281 : : {
282 : 598193 : CopySendString(cstate, cstate->opts.null_print_client);
283 : : }
284 : : else
285 : : {
286 : : char *string;
287 : :
288 : 6489894 : string = OutputFunctionCall(&out_functions[attnum - 1],
289 : : value);
290 : :
291 [ + + ]: 6489894 : if (is_csv)
292 : 297 : CopyAttributeOutCSV(cstate, string,
293 : 297 : cstate->opts.force_quote_flags[attnum - 1]);
294 : : else
295 : 6489597 : CopyAttributeOutText(cstate, string);
296 : : }
297 : : }
298 : :
299 : 1829756 : CopySendTextLikeEndOfRow(cstate);
300 : 1829756 : }
301 : :
302 : : /* Implementation of the end callback for text and CSV formats */
303 : : static void
304 : 4816 : CopyToTextLikeEnd(CopyToState cstate)
305 : : {
306 : : /* Nothing to do here */
307 : 4816 : }
308 : :
309 : : /*
310 : : * Implementation of the start callback for binary format. Send a header
311 : : * for a binary copy.
312 : : */
313 : : static void
314 : 7 : CopyToBinaryStart(CopyToState cstate, TupleDesc tupDesc)
315 : : {
316 : : int32 tmp;
317 : :
318 : : /* Signature */
319 : 7 : CopySendData(cstate, BinarySignature, 11);
320 : : /* Flags field */
321 : 7 : tmp = 0;
322 : 7 : CopySendInt32(cstate, tmp);
323 : : /* No header extension */
324 : 7 : tmp = 0;
325 : 7 : CopySendInt32(cstate, tmp);
326 : 7 : }
327 : :
328 : : /*
329 : : * Implementation of the outfunc callback for binary format. Assign
330 : : * the binary output function to the given *finfo.
331 : : */
332 : : static void
333 : 31 : CopyToBinaryOutFunc(CopyToState cstate, Oid atttypid, FmgrInfo *finfo)
334 : : {
335 : : Oid func_oid;
336 : : bool is_varlena;
337 : :
338 : : /* Set output function for an attribute */
339 : 31 : getTypeBinaryOutputInfo(atttypid, &func_oid, &is_varlena);
340 : 30 : fmgr_info(func_oid, finfo);
341 : 30 : }
342 : :
343 : : /* Implementation of the per-row callback for binary format */
344 : : static void
345 : 16 : CopyToBinaryOneRow(CopyToState cstate, TupleTableSlot *slot)
346 : : {
347 : 16 : FmgrInfo *out_functions = cstate->out_functions;
348 : :
349 : : /* Binary per-tuple header */
350 : 16 : CopySendInt16(cstate, list_length(cstate->attnumlist));
351 : :
352 [ + - + + : 112 : foreach_int(attnum, cstate->attnumlist)
+ + ]
353 : : {
354 : 80 : Datum value = slot->tts_values[attnum - 1];
355 : 80 : bool isnull = slot->tts_isnull[attnum - 1];
356 : :
357 [ + + ]: 80 : if (isnull)
358 : : {
359 : 15 : CopySendInt32(cstate, -1);
360 : : }
361 : : else
362 : : {
363 : : bytea *outputbytes;
364 : :
365 : 65 : outputbytes = SendFunctionCall(&out_functions[attnum - 1],
366 : : value);
367 : 65 : CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
368 : 65 : CopySendData(cstate, VARDATA(outputbytes),
369 : 65 : VARSIZE(outputbytes) - VARHDRSZ);
370 : : }
371 : : }
372 : :
373 : 16 : CopySendEndOfRow(cstate);
374 : 16 : }
375 : :
376 : : /* Implementation of the end callback for binary format */
377 : : static void
378 : 7 : CopyToBinaryEnd(CopyToState cstate)
379 : : {
380 : : /* Generate trailer for a binary copy */
381 : 7 : CopySendInt16(cstate, -1);
382 : : /* Need to flush out the trailer */
383 : 7 : CopySendEndOfRow(cstate);
384 : 7 : }
385 : :
386 : : /*
387 : : * Send copy start/stop messages for frontend copies. These have changed
388 : : * in past protocol redesigns.
389 : : */
390 : : static void
1748 heikki.linnakangas@i 391 : 4792 : SendCopyBegin(CopyToState cstate)
392 : : {
393 : : StringInfoData buf;
1647 394 : 4792 : int natts = list_length(cstate->attnumlist);
395 : 4792 : int16 format = (cstate->opts.binary ? 1 : 0);
396 : : int i;
397 : :
746 nathan@postgresql.or 398 : 4792 : pq_beginmessage(&buf, PqMsg_CopyOutResponse);
1647 heikki.linnakangas@i 399 : 4792 : pq_sendbyte(&buf, format); /* overall format */
400 : 4792 : pq_sendint16(&buf, natts);
401 [ + + ]: 21023 : for (i = 0; i < natts; i++)
402 : 16231 : pq_sendint16(&buf, format); /* per-column formats */
403 : 4792 : pq_endmessage(&buf);
404 : 4792 : cstate->copy_dest = COPY_FRONTEND;
1748 405 : 4792 : }
406 : :
407 : : static void
408 : 4791 : SendCopyEnd(CopyToState cstate)
409 : : {
410 : : /* Shouldn't have any unsent data */
1647 411 [ - + ]: 4791 : Assert(cstate->fe_msgbuf->len == 0);
412 : : /* Send Copy Done message */
746 nathan@postgresql.or 413 : 4791 : pq_putemptymessage(PqMsg_CopyDone);
1748 heikki.linnakangas@i 414 : 4791 : }
415 : :
416 : : /*----------
417 : : * CopySendData sends output data to the destination (file or frontend)
418 : : * CopySendString does the same for null-terminated strings
419 : : * CopySendChar does the same for single characters
420 : : * CopySendEndOfRow does the appropriate thing at end of each data row
421 : : * (data is not actually flushed except by CopySendEndOfRow)
422 : : *
423 : : * NB: no data conversion is applied by these functions
424 : : *----------
425 : : */
426 : : static void
427 : 6394086 : CopySendData(CopyToState cstate, const void *databuf, int datasize)
428 : : {
429 : 6394086 : appendBinaryStringInfo(cstate->fe_msgbuf, databuf, datasize);
430 : 6394086 : }
431 : :
432 : : static void
433 : 598340 : CopySendString(CopyToState cstate, const char *str)
434 : : {
435 : 598340 : appendBinaryStringInfo(cstate->fe_msgbuf, str, strlen(str));
436 : 598340 : }
437 : :
438 : : static void
439 : 7108241 : CopySendChar(CopyToState cstate, char c)
440 : : {
441 [ + + ]: 7108241 : appendStringInfoCharMacro(cstate->fe_msgbuf, c);
442 : 7108241 : }
443 : :
444 : : static void
445 : 1829794 : CopySendEndOfRow(CopyToState cstate)
446 : : {
447 : 1829794 : StringInfo fe_msgbuf = cstate->fe_msgbuf;
448 : :
449 [ + + + - ]: 1829794 : switch (cstate->copy_dest)
450 : : {
451 : 6141 : case COPY_FILE:
452 [ + - ]: 6141 : if (fwrite(fe_msgbuf->data, fe_msgbuf->len, 1,
453 [ - + ]: 6141 : cstate->copy_file) != 1 ||
454 : 6141 : ferror(cstate->copy_file))
455 : : {
1748 heikki.linnakangas@i 456 [ # # ]:UBC 0 : if (cstate->is_program)
457 : : {
458 [ # # ]: 0 : if (errno == EPIPE)
459 : : {
460 : : /*
461 : : * The pipe will be closed automatically on error at
462 : : * the end of transaction, but we might get a better
463 : : * error message from the subprocess' exit code than
464 : : * just "Broken Pipe"
465 : : */
466 : 0 : ClosePipeToProgram(cstate);
467 : :
468 : : /*
469 : : * If ClosePipeToProgram() didn't throw an error, the
470 : : * program terminated normally, but closed the pipe
471 : : * first. Restore errno, and throw an error.
472 : : */
473 : 0 : errno = EPIPE;
474 : : }
475 [ # # ]: 0 : ereport(ERROR,
476 : : (errcode_for_file_access(),
477 : : errmsg("could not write to COPY program: %m")));
478 : : }
479 : : else
480 [ # # ]: 0 : ereport(ERROR,
481 : : (errcode_for_file_access(),
482 : : errmsg("could not write to COPY file: %m")));
483 : : }
1748 heikki.linnakangas@i 484 :CBC 6141 : break;
1647 485 : 1823650 : case COPY_FRONTEND:
486 : : /* Dump the accumulated row as one CopyData message */
746 nathan@postgresql.or 487 : 1823650 : (void) pq_putmessage(PqMsg_CopyData, fe_msgbuf->data, fe_msgbuf->len);
1748 heikki.linnakangas@i 488 : 1823650 : break;
1061 michael@paquier.xyz 489 : 3 : case COPY_CALLBACK:
490 : 3 : cstate->data_dest_cb(fe_msgbuf->data, fe_msgbuf->len);
491 : 3 : break;
492 : : }
493 : :
494 : : /* Update the progress */
1704 tomas.vondra@postgre 495 : 1829794 : cstate->bytes_processed += fe_msgbuf->len;
496 : 1829794 : pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, cstate->bytes_processed);
497 : :
1748 heikki.linnakangas@i 498 : 1829794 : resetStringInfo(fe_msgbuf);
499 : 1829794 : }
500 : :
501 : : /*
502 : : * Wrapper function of CopySendEndOfRow for text and CSV formats. Sends the
503 : : * line termination and do common appropriate things for the end of row.
504 : : */
505 : : static inline void
191 msawada@postgresql.o 506 : 1829771 : CopySendTextLikeEndOfRow(CopyToState cstate)
507 : : {
508 [ + + + ]: 1829771 : switch (cstate->copy_dest)
509 : : {
510 : 6129 : case COPY_FILE:
511 : : /* Default line termination depends on platform */
512 : : #ifndef WIN32
513 : 6129 : CopySendChar(cstate, '\n');
514 : : #else
515 : : CopySendString(cstate, "\r\n");
516 : : #endif
517 : 6129 : break;
518 : 1823639 : case COPY_FRONTEND:
519 : : /* The FE/BE protocol uses \n as newline for all platforms */
520 : 1823639 : CopySendChar(cstate, '\n');
521 : 1823639 : break;
522 : 3 : default:
523 : 3 : break;
524 : : }
525 : :
526 : : /* Now take the actions related to the end of a row */
527 : 1829771 : CopySendEndOfRow(cstate);
528 : 1829771 : }
529 : :
530 : : /*
531 : : * These functions do apply some data conversion
532 : : */
533 : :
534 : : /*
535 : : * CopySendInt32 sends an int32 in network byte order
536 : : */
537 : : static inline void
1748 heikki.linnakangas@i 538 : 94 : CopySendInt32(CopyToState cstate, int32 val)
539 : : {
540 : : uint32 buf;
541 : :
542 : 94 : buf = pg_hton32((uint32) val);
543 : 94 : CopySendData(cstate, &buf, sizeof(buf));
544 : 94 : }
545 : :
546 : : /*
547 : : * CopySendInt16 sends an int16 in network byte order
548 : : */
549 : : static inline void
550 : 23 : CopySendInt16(CopyToState cstate, int16 val)
551 : : {
552 : : uint16 buf;
553 : :
554 : 23 : buf = pg_hton16((uint16) val);
555 : 23 : CopySendData(cstate, &buf, sizeof(buf));
556 : 23 : }
557 : :
558 : : /*
559 : : * Closes the pipe to an external program, checking the pclose() return code.
560 : : */
561 : : static void
1748 heikki.linnakangas@i 562 :UBC 0 : ClosePipeToProgram(CopyToState cstate)
563 : : {
564 : : int pclose_rc;
565 : :
566 [ # # ]: 0 : Assert(cstate->is_program);
567 : :
568 : 0 : pclose_rc = ClosePipeStream(cstate->copy_file);
569 [ # # ]: 0 : if (pclose_rc == -1)
570 [ # # ]: 0 : ereport(ERROR,
571 : : (errcode_for_file_access(),
572 : : errmsg("could not close pipe to external command: %m")));
573 [ # # ]: 0 : else if (pclose_rc != 0)
574 : : {
575 [ # # ]: 0 : ereport(ERROR,
576 : : (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
577 : : errmsg("program \"%s\" failed",
578 : : cstate->filename),
579 : : errdetail_internal("%s", wait_result_to_str(pclose_rc))));
580 : : }
581 : 0 : }
582 : :
583 : : /*
584 : : * Release resources allocated in a cstate for COPY TO/FROM.
585 : : */
586 : : static void
1748 heikki.linnakangas@i 587 :CBC 4823 : EndCopy(CopyToState cstate)
588 : : {
589 [ - + ]: 4823 : if (cstate->is_program)
590 : : {
1748 heikki.linnakangas@i 591 :UBC 0 : ClosePipeToProgram(cstate);
592 : : }
593 : : else
594 : : {
1748 heikki.linnakangas@i 595 [ + + - + ]:CBC 4823 : if (cstate->filename != NULL && FreeFile(cstate->copy_file))
1748 heikki.linnakangas@i 596 [ # # ]:UBC 0 : ereport(ERROR,
597 : : (errcode_for_file_access(),
598 : : errmsg("could not close file \"%s\": %m",
599 : : cstate->filename)));
600 : : }
601 : :
1704 tomas.vondra@postgre 602 :CBC 4823 : pgstat_progress_end_command();
603 : :
1748 heikki.linnakangas@i 604 : 4823 : MemoryContextDelete(cstate->copycontext);
605 : 4823 : pfree(cstate);
606 : 4823 : }
607 : :
608 : : /*
609 : : * Setup CopyToState to read tuples from a table or a query for COPY TO.
610 : : *
611 : : * 'rel': Relation to be copied
612 : : * 'raw_query': Query whose results are to be copied
613 : : * 'queryRelId': OID of base relation to convert to a query (for RLS)
614 : : * 'filename': Name of server-local file to write, NULL for STDOUT
615 : : * 'is_program': true if 'filename' is program to execute
616 : : * 'data_dest_cb': Callback that processes the output data
617 : : * 'attnamelist': List of char *, columns to include. NIL selects all cols.
618 : : * 'options': List of DefElem. See copy_opt_item in gram.y for selections.
619 : : *
620 : : * Returns a CopyToState, to be passed to DoCopyTo() and related functions.
621 : : */
622 : : CopyToState
623 : 4924 : BeginCopyTo(ParseState *pstate,
624 : : Relation rel,
625 : : RawStmt *raw_query,
626 : : Oid queryRelId,
627 : : const char *filename,
628 : : bool is_program,
629 : : copy_data_dest_cb data_dest_cb,
630 : : List *attnamelist,
631 : : List *options)
632 : : {
633 : : CopyToState cstate;
1061 michael@paquier.xyz 634 [ + + + + ]: 4924 : bool pipe = (filename == NULL && data_dest_cb == NULL);
635 : : TupleDesc tupDesc;
636 : : int num_phys_attrs;
637 : : MemoryContext oldcontext;
1642 638 : 4924 : const int progress_cols[] = {
639 : : PROGRESS_COPY_COMMAND,
640 : : PROGRESS_COPY_TYPE
641 : : };
642 : 4924 : int64 progress_vals[] = {
643 : : PROGRESS_COPY_COMMAND_TO,
644 : : 0
645 : : };
646 : :
1748 heikki.linnakangas@i 647 [ + + + + ]: 4924 : if (rel != NULL && rel->rd_rel->relkind != RELKIND_RELATION)
648 : : {
649 [ + + ]: 12 : if (rel->rd_rel->relkind == RELKIND_VIEW)
650 [ + - ]: 6 : ereport(ERROR,
651 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
652 : : errmsg("cannot copy from view \"%s\"",
653 : : RelationGetRelationName(rel)),
654 : : errhint("Try the COPY (SELECT ...) TO variant.")));
655 [ + - ]: 6 : else if (rel->rd_rel->relkind == RELKIND_MATVIEW)
656 : : {
155 fujii@postgresql.org 657 [ + + ]: 6 : if (!RelationIsPopulated(rel))
658 [ + - ]: 3 : ereport(ERROR,
659 : : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
660 : : errmsg("cannot copy from unpopulated materialized view \"%s\"",
661 : : RelationGetRelationName(rel)),
662 : : errhint("Use the REFRESH MATERIALIZED VIEW command."));
663 : : }
1748 heikki.linnakangas@i 664 [ # # ]:UBC 0 : else if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
665 [ # # ]: 0 : ereport(ERROR,
666 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
667 : : errmsg("cannot copy from foreign table \"%s\"",
668 : : RelationGetRelationName(rel)),
669 : : errhint("Try the COPY (SELECT ...) TO variant.")));
670 [ # # ]: 0 : else if (rel->rd_rel->relkind == RELKIND_SEQUENCE)
671 [ # # ]: 0 : ereport(ERROR,
672 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
673 : : errmsg("cannot copy from sequence \"%s\"",
674 : : RelationGetRelationName(rel))));
675 [ # # ]: 0 : else if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
676 [ # # ]: 0 : ereport(ERROR,
677 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
678 : : errmsg("cannot copy from partitioned table \"%s\"",
679 : : RelationGetRelationName(rel)),
680 : : errhint("Try the COPY (SELECT ...) TO variant.")));
681 : : else
682 [ # # ]: 0 : ereport(ERROR,
683 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
684 : : errmsg("cannot copy from non-table relation \"%s\"",
685 : : RelationGetRelationName(rel))));
686 : : }
687 : :
688 : :
689 : : /* Allocate workspace and zero all fields */
1748 heikki.linnakangas@i 690 :CBC 4915 : cstate = (CopyToStateData *) palloc0(sizeof(CopyToStateData));
691 : :
692 : : /*
693 : : * We allocate everything used by a cstate in a new memory context. This
694 : : * avoids memory leaks during repeated use of COPY in a query.
695 : : */
696 : 4915 : cstate->copycontext = AllocSetContextCreate(CurrentMemoryContext,
697 : : "COPY",
698 : : ALLOCSET_DEFAULT_SIZES);
699 : :
700 : 4915 : oldcontext = MemoryContextSwitchTo(cstate->copycontext);
701 : :
702 : : /* Extract options from the statement node tree */
1578 tgl@sss.pgh.pa.us 703 : 4915 : ProcessCopyOptions(pstate, &cstate->opts, false /* is_from */ , options);
704 : :
705 : : /* Set format routine */
186 msawada@postgresql.o 706 : 4891 : cstate->routine = CopyToGetRoutine(&cstate->opts);
707 : :
708 : : /* Process the source/target relation or query */
1748 heikki.linnakangas@i 709 [ + + ]: 4891 : if (rel)
710 : : {
711 [ - + ]: 4602 : Assert(!raw_query);
712 : :
713 : 4602 : cstate->rel = rel;
714 : :
715 : 4602 : tupDesc = RelationGetDescr(cstate->rel);
716 : : }
717 : : else
718 : : {
719 : : List *rewritten;
720 : : Query *query;
721 : : PlannedStmt *plan;
722 : : DestReceiver *dest;
723 : :
724 : 289 : cstate->rel = NULL;
725 : :
726 : : /*
727 : : * Run parse analysis and rewrite. Note this also acquires sufficient
728 : : * locks on the source table(s).
729 : : */
1282 peter@eisentraut.org 730 : 289 : rewritten = pg_analyze_and_rewrite_fixedparams(raw_query,
731 : : pstate->p_sourcetext, NULL, 0,
732 : : NULL);
733 : :
734 : : /* check that we got back something we can work with */
1748 heikki.linnakangas@i 735 [ + + ]: 283 : if (rewritten == NIL)
736 : : {
737 [ + - ]: 9 : ereport(ERROR,
738 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
739 : : errmsg("DO INSTEAD NOTHING rules are not supported for COPY")));
740 : : }
741 [ + + ]: 274 : else if (list_length(rewritten) > 1)
742 : : {
743 : : ListCell *lc;
744 : :
745 : : /* examine queries to determine which error message to issue */
746 [ + - + + : 51 : foreach(lc, rewritten)
+ + ]
747 : : {
748 : 42 : Query *q = lfirst_node(Query, lc);
749 : :
750 [ + + ]: 42 : if (q->querySource == QSRC_QUAL_INSTEAD_RULE)
751 [ + - ]: 9 : ereport(ERROR,
752 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
753 : : errmsg("conditional DO INSTEAD rules are not supported for COPY")));
754 [ + + ]: 33 : if (q->querySource == QSRC_NON_INSTEAD_RULE)
755 [ + - ]: 9 : ereport(ERROR,
756 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
757 : : errmsg("DO ALSO rules are not supported for COPY")));
758 : : }
759 : :
760 [ + - ]: 9 : ereport(ERROR,
761 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
762 : : errmsg("multi-statement DO INSTEAD rules are not supported for COPY")));
763 : : }
764 : :
765 : 247 : query = linitial_node(Query, rewritten);
766 : :
767 : : /* The grammar allows SELECT INTO, but we don't support that */
768 [ + + ]: 247 : if (query->utilityStmt != NULL &&
769 [ + + ]: 9 : IsA(query->utilityStmt, CreateTableAsStmt))
770 [ + - ]: 6 : ereport(ERROR,
771 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
772 : : errmsg("COPY (SELECT INTO) is not supported")));
773 : :
774 : : /* The only other utility command we could see is NOTIFY */
320 tgl@sss.pgh.pa.us 775 [ + + ]: 241 : if (query->utilityStmt != NULL)
776 [ + - ]: 3 : ereport(ERROR,
777 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
778 : : errmsg("COPY query must not be a utility command")));
779 : :
780 : : /*
781 : : * Similarly the grammar doesn't enforce the presence of a RETURNING
782 : : * clause, but this is required here.
783 : : */
1748 heikki.linnakangas@i 784 [ + + ]: 238 : if (query->commandType != CMD_SELECT &&
785 [ + + ]: 55 : query->returningList == NIL)
786 : : {
787 [ + + + + : 12 : Assert(query->commandType == CMD_INSERT ||
+ + - + ]
788 : : query->commandType == CMD_UPDATE ||
789 : : query->commandType == CMD_DELETE ||
790 : : query->commandType == CMD_MERGE);
791 : :
792 [ + - ]: 12 : ereport(ERROR,
793 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
794 : : errmsg("COPY query must have a RETURNING clause")));
795 : : }
796 : :
797 : : /* plan the query */
798 : 226 : plan = pg_plan_query(query, pstate->p_sourcetext,
799 : : CURSOR_OPT_PARALLEL_OK, NULL);
800 : :
801 : : /*
802 : : * With row-level security and a user using "COPY relation TO", we
803 : : * have to convert the "COPY relation TO" to a query-based COPY (eg:
804 : : * "COPY (SELECT * FROM ONLY relation) TO"), to allow the rewriter to
805 : : * add in any RLS clauses.
806 : : *
807 : : * When this happens, we are passed in the relid of the originally
808 : : * found relation (which we have locked). As the planner will look up
809 : : * the relation again, we double-check here to make sure it found the
810 : : * same one that we have locked.
811 : : */
812 [ + + ]: 225 : if (queryRelId != InvalidOid)
813 : : {
814 : : /*
815 : : * Note that with RLS involved there may be multiple relations,
816 : : * and while the one we need is almost certainly first, we don't
817 : : * make any guarantees of that in the planner, so check the whole
818 : : * list and make sure we find the original relation.
819 : : */
820 [ - + ]: 27 : if (!list_member_oid(plan->relationOids, queryRelId))
1748 heikki.linnakangas@i 821 [ # # ]:UBC 0 : ereport(ERROR,
822 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
823 : : errmsg("relation referenced by COPY statement has changed")));
824 : : }
825 : :
826 : : /*
827 : : * Use a snapshot with an updated command ID to ensure this query sees
828 : : * results of any previously executed queries.
829 : : */
1748 heikki.linnakangas@i 830 :CBC 225 : PushCopiedSnapshot(GetActiveSnapshot());
831 : 225 : UpdateActiveSnapshotCommandId();
832 : :
833 : : /* Create dest receiver for COPY OUT */
834 : 225 : dest = CreateDestReceiver(DestCopyOut);
835 : 225 : ((DR_copy *) dest)->cstate = cstate;
836 : :
837 : : /* Create a QueryDesc requesting no output */
107 amitlan@postgresql.o 838 : 225 : cstate->queryDesc = CreateQueryDesc(plan, pstate->p_sourcetext,
839 : : GetActiveSnapshot(),
840 : : InvalidSnapshot,
841 : : dest, NULL, NULL, 0);
842 : :
843 : : /*
844 : : * Call ExecutorStart to prepare the plan for execution.
845 : : *
846 : : * ExecutorStart computes a result tupdesc for us
847 : : */
848 : 225 : ExecutorStart(cstate->queryDesc, 0);
849 : :
1748 heikki.linnakangas@i 850 : 222 : tupDesc = cstate->queryDesc->tupDesc;
851 : : }
852 : :
853 : : /* Generate or convert list of attributes to process */
854 : 4824 : cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
855 : :
856 : 4824 : num_phys_attrs = tupDesc->natts;
857 : :
858 : : /* Convert FORCE_QUOTE name list to per-column flags, check validity */
859 : 4824 : cstate->opts.force_quote_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
860 [ + + ]: 4824 : if (cstate->opts.force_quote_all)
861 : : {
707 andrew@dunslane.net 862 [ + - - + : 9 : MemSet(cstate->opts.force_quote_flags, true, num_phys_attrs * sizeof(bool));
- - - - -
- ]
863 : : }
1748 heikki.linnakangas@i 864 [ + + ]: 4815 : else if (cstate->opts.force_quote)
865 : : {
866 : : List *attnums;
867 : : ListCell *cur;
868 : :
869 : 12 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_quote);
870 : :
871 [ + - + + : 24 : foreach(cur, attnums)
+ + ]
872 : : {
873 : 12 : int attnum = lfirst_int(cur);
874 : 12 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
875 : :
876 [ - + ]: 12 : if (!list_member_int(cstate->attnumlist, attnum))
1748 heikki.linnakangas@i 877 [ # # ]:UBC 0 : ereport(ERROR,
878 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
879 : : /*- translator: %s is the name of a COPY option, e.g. FORCE_NOT_NULL */
880 : : errmsg("%s column \"%s\" not referenced by COPY",
881 : : "FORCE_QUOTE", NameStr(attr->attname))));
1748 heikki.linnakangas@i 882 :CBC 12 : cstate->opts.force_quote_flags[attnum - 1] = true;
883 : : }
884 : : }
885 : :
886 : : /* Use client encoding when ENCODING option is not specified. */
887 [ + + ]: 4824 : if (cstate->opts.file_encoding < 0)
888 : 4815 : cstate->file_encoding = pg_get_client_encoding();
889 : : else
890 : 9 : cstate->file_encoding = cstate->opts.file_encoding;
891 : :
892 : : /*
893 : : * Set up encoding conversion info if the file and server encodings differ
894 : : * (see also pg_server_to_any).
895 : : */
575 michael@paquier.xyz 896 [ + + ]: 4824 : if (cstate->file_encoding == GetDatabaseEncoding() ||
897 [ + + ]: 4 : cstate->file_encoding == PG_SQL_ASCII)
898 : 4823 : cstate->need_transcoding = false;
899 : : else
900 : 1 : cstate->need_transcoding = true;
901 : :
902 : : /* See Multibyte encoding comment above */
1748 heikki.linnakangas@i 903 [ - + - - ]: 4824 : cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding);
904 : :
905 : 4824 : cstate->copy_dest = COPY_FILE; /* default */
906 : :
1061 michael@paquier.xyz 907 [ + + ]: 4824 : if (data_dest_cb)
908 : : {
909 : 1 : progress_vals[1] = PROGRESS_COPY_TYPE_CALLBACK;
910 : 1 : cstate->copy_dest = COPY_CALLBACK;
911 : 1 : cstate->data_dest_cb = data_dest_cb;
912 : : }
913 [ + + ]: 4823 : else if (pipe)
914 : : {
1642 915 : 4792 : progress_vals[1] = PROGRESS_COPY_TYPE_PIPE;
916 : :
1748 heikki.linnakangas@i 917 [ - + ]: 4792 : Assert(!is_program); /* the grammar does not allow this */
918 [ - + ]: 4792 : if (whereToSendOutput != DestRemote)
1748 heikki.linnakangas@i 919 :UBC 0 : cstate->copy_file = stdout;
920 : : }
921 : : else
922 : : {
1748 heikki.linnakangas@i 923 :CBC 31 : cstate->filename = pstrdup(filename);
924 : 31 : cstate->is_program = is_program;
925 : :
926 [ - + ]: 31 : if (is_program)
927 : : {
1642 michael@paquier.xyz 928 :UBC 0 : progress_vals[1] = PROGRESS_COPY_TYPE_PROGRAM;
1748 heikki.linnakangas@i 929 : 0 : cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_W);
930 [ # # ]: 0 : if (cstate->copy_file == NULL)
931 [ # # ]: 0 : ereport(ERROR,
932 : : (errcode_for_file_access(),
933 : : errmsg("could not execute command \"%s\": %m",
934 : : cstate->filename)));
935 : : }
936 : : else
937 : : {
938 : : mode_t oumask; /* Pre-existing umask value */
939 : : struct stat st;
940 : :
1642 michael@paquier.xyz 941 :CBC 31 : progress_vals[1] = PROGRESS_COPY_TYPE_FILE;
942 : :
943 : : /*
944 : : * Prevent write to relative path ... too easy to shoot oneself in
945 : : * the foot by overwriting a database file ...
946 : : */
1748 heikki.linnakangas@i 947 [ - + ]: 31 : if (!is_absolute_path(filename))
1748 heikki.linnakangas@i 948 [ # # ]:UBC 0 : ereport(ERROR,
949 : : (errcode(ERRCODE_INVALID_NAME),
950 : : errmsg("relative path not allowed for COPY to file")));
951 : :
1748 heikki.linnakangas@i 952 :CBC 31 : oumask = umask(S_IWGRP | S_IWOTH);
953 [ + - ]: 31 : PG_TRY();
954 : : {
955 : 31 : cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_W);
956 : : }
1748 heikki.linnakangas@i 957 :UBC 0 : PG_FINALLY();
958 : : {
1748 heikki.linnakangas@i 959 :CBC 31 : umask(oumask);
960 : : }
961 [ - + ]: 31 : PG_END_TRY();
962 [ - + ]: 31 : if (cstate->copy_file == NULL)
963 : : {
964 : : /* copy errno because ereport subfunctions might change it */
1748 heikki.linnakangas@i 965 :UBC 0 : int save_errno = errno;
966 : :
967 [ # # # # : 0 : ereport(ERROR,
# # ]
968 : : (errcode_for_file_access(),
969 : : errmsg("could not open file \"%s\" for writing: %m",
970 : : cstate->filename),
971 : : (save_errno == ENOENT || save_errno == EACCES) ?
972 : : errhint("COPY TO instructs the PostgreSQL server process to write a file. "
973 : : "You may want a client-side facility such as psql's \\copy.") : 0));
974 : : }
975 : :
1748 heikki.linnakangas@i 976 [ - + ]:CBC 31 : if (fstat(fileno(cstate->copy_file), &st))
1748 heikki.linnakangas@i 977 [ # # ]:UBC 0 : ereport(ERROR,
978 : : (errcode_for_file_access(),
979 : : errmsg("could not stat file \"%s\": %m",
980 : : cstate->filename)));
981 : :
1748 heikki.linnakangas@i 982 [ - + ]:CBC 31 : if (S_ISDIR(st.st_mode))
1748 heikki.linnakangas@i 983 [ # # ]:UBC 0 : ereport(ERROR,
984 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
985 : : errmsg("\"%s\" is a directory", cstate->filename)));
986 : : }
987 : : }
988 : :
989 : : /* initialize progress */
1704 tomas.vondra@postgre 990 :CBC 4824 : pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
991 [ + + ]: 4824 : cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
1642 michael@paquier.xyz 992 : 4824 : pgstat_progress_update_multi_param(2, progress_cols, progress_vals);
993 : :
1704 tomas.vondra@postgre 994 : 4824 : cstate->bytes_processed = 0;
995 : :
1748 heikki.linnakangas@i 996 : 4824 : MemoryContextSwitchTo(oldcontext);
997 : :
998 : 4824 : return cstate;
999 : : }
1000 : :
1001 : : /*
1002 : : * Clean up storage and release resources for COPY TO.
1003 : : */
1004 : : void
1005 : 4823 : EndCopyTo(CopyToState cstate)
1006 : : {
1007 [ + + ]: 4823 : if (cstate->queryDesc != NULL)
1008 : : {
1009 : : /* Close down the query and free resources. */
1010 : 222 : ExecutorFinish(cstate->queryDesc);
1011 : 222 : ExecutorEnd(cstate->queryDesc);
1012 : 222 : FreeQueryDesc(cstate->queryDesc);
1013 : 222 : PopActiveSnapshot();
1014 : : }
1015 : :
1016 : : /* Clean up storage */
1017 : 4823 : EndCopy(cstate);
1018 : 4823 : }
1019 : :
1020 : : /*
1021 : : * Copy from relation or query TO file.
1022 : : *
1023 : : * Returns the number of rows processed.
1024 : : */
1025 : : uint64
1647 1026 : 4824 : DoCopyTo(CopyToState cstate)
1027 : : {
1061 michael@paquier.xyz 1028 [ + + + + ]: 4824 : bool pipe = (cstate->filename == NULL && cstate->data_dest_cb == NULL);
1647 heikki.linnakangas@i 1029 [ + + + - ]: 4824 : bool fe_copy = (pipe && whereToSendOutput == DestRemote);
1030 : : TupleDesc tupDesc;
1031 : : int num_phys_attrs;
1032 : : ListCell *cur;
1033 : : uint64 processed;
1034 : :
1035 [ + + ]: 4824 : if (fe_copy)
1036 : 4792 : SendCopyBegin(cstate);
1037 : :
1748 1038 [ + + ]: 4824 : if (cstate->rel)
1039 : 4602 : tupDesc = RelationGetDescr(cstate->rel);
1040 : : else
1041 : 222 : tupDesc = cstate->queryDesc->tupDesc;
1042 : 4824 : num_phys_attrs = tupDesc->natts;
1578 tgl@sss.pgh.pa.us 1043 : 4824 : cstate->opts.null_print_client = cstate->opts.null_print; /* default */
1044 : :
1045 : : /* We use fe_msgbuf as a per-row buffer regardless of copy_dest */
1748 heikki.linnakangas@i 1046 : 4824 : cstate->fe_msgbuf = makeStringInfo();
1047 : :
1048 : : /* Get info about the columns we need to process. */
1049 : 4824 : cstate->out_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
1050 [ + + + + : 21170 : foreach(cur, cstate->attnumlist)
+ + ]
1051 : : {
1052 : 16347 : int attnum = lfirst_int(cur);
1053 : 16347 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1054 : :
191 msawada@postgresql.o 1055 : 16347 : cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
1056 : 16347 : &cstate->out_functions[attnum - 1]);
1057 : : }
1058 : :
1059 : : /*
1060 : : * Create a temporary memory context that we can reset once per row to
1061 : : * recover palloc'd memory. This avoids any problems with leaks inside
1062 : : * datatype output routines, and should be faster than retail pfree's
1063 : : * anyway. (We don't need a whole econtext as CopyFrom does.)
1064 : : */
1748 heikki.linnakangas@i 1065 : 4823 : cstate->rowcontext = AllocSetContextCreate(CurrentMemoryContext,
1066 : : "COPY TO",
1067 : : ALLOCSET_DEFAULT_SIZES);
1068 : :
191 msawada@postgresql.o 1069 : 4823 : cstate->routine->CopyToStart(cstate, tupDesc);
1070 : :
1748 heikki.linnakangas@i 1071 [ + + ]: 4823 : if (cstate->rel)
1072 : : {
1073 : : TupleTableSlot *slot;
1074 : : TableScanDesc scandesc;
1075 : :
1076 : 4601 : scandesc = table_beginscan(cstate->rel, GetActiveSnapshot(), 0, NULL);
1077 : 4601 : slot = table_slot_create(cstate->rel, NULL);
1078 : :
1079 : 4601 : processed = 0;
1080 [ + + ]: 1830871 : while (table_scan_getnextslot(scandesc, ForwardScanDirection, slot))
1081 : : {
1082 [ - + ]: 1826270 : CHECK_FOR_INTERRUPTS();
1083 : :
1084 : : /* Deconstruct the tuple ... */
1085 : 1826270 : slot_getallattrs(slot);
1086 : :
1087 : : /* Format and send the data */
1088 : 1826270 : CopyOneRowTo(cstate, slot);
1089 : :
1090 : : /*
1091 : : * Increment the number of processed tuples, and report the
1092 : : * progress.
1093 : : */
1642 michael@paquier.xyz 1094 : 1826270 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
1095 : : ++processed);
1096 : : }
1097 : :
1748 heikki.linnakangas@i 1098 : 4601 : ExecDropSingleTupleTableSlot(slot);
1099 : 4601 : table_endscan(scandesc);
1100 : : }
1101 : : else
1102 : : {
1103 : : /* run the plan --- the dest receiver will send tuples */
271 tgl@sss.pgh.pa.us 1104 : 222 : ExecutorRun(cstate->queryDesc, ForwardScanDirection, 0);
1748 heikki.linnakangas@i 1105 : 222 : processed = ((DR_copy *) cstate->queryDesc->dest)->processed;
1106 : : }
1107 : :
191 msawada@postgresql.o 1108 : 4823 : cstate->routine->CopyToEnd(cstate);
1109 : :
1748 heikki.linnakangas@i 1110 : 4823 : MemoryContextDelete(cstate->rowcontext);
1111 : :
1647 1112 [ + + ]: 4823 : if (fe_copy)
1113 : 4791 : SendCopyEnd(cstate);
1114 : :
1748 1115 : 4823 : return processed;
1116 : : }
1117 : :
1118 : : /*
1119 : : * Emit one row during DoCopyTo().
1120 : : */
1121 : : static inline void
1122 : 1829772 : CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
1123 : : {
1124 : : MemoryContext oldcontext;
1125 : :
1126 : 1829772 : MemoryContextReset(cstate->rowcontext);
1127 : 1829772 : oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
1128 : :
1129 : : /* Make sure the tuple is fully deconstructed */
1130 : 1829772 : slot_getallattrs(slot);
1131 : :
191 msawada@postgresql.o 1132 : 1829772 : cstate->routine->CopyToOneRow(cstate, slot);
1133 : :
1748 heikki.linnakangas@i 1134 : 1829772 : MemoryContextSwitchTo(oldcontext);
1135 : 1829772 : }
1136 : :
1137 : : /*
1138 : : * Send text representation of one attribute, with conversion and escaping
1139 : : */
1140 : : #define DUMPSOFAR() \
1141 : : do { \
1142 : : if (ptr > start) \
1143 : : CopySendData(cstate, start, ptr - start); \
1144 : : } while (0)
1145 : :
1146 : : static void
577 michael@paquier.xyz 1147 : 6489609 : CopyAttributeOutText(CopyToState cstate, const char *string)
1148 : : {
1149 : : const char *ptr;
1150 : : const char *start;
1151 : : char c;
1748 heikki.linnakangas@i 1152 : 6489609 : char delimc = cstate->opts.delim[0];
1153 : :
1154 [ - + ]: 6489609 : if (cstate->need_transcoding)
1748 heikki.linnakangas@i 1155 :UBC 0 : ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding);
1156 : : else
1748 heikki.linnakangas@i 1157 :CBC 6489609 : ptr = string;
1158 : :
1159 : : /*
1160 : : * We have to grovel through the string searching for control characters
1161 : : * and instances of the delimiter character. In most cases, though, these
1162 : : * are infrequent. To avoid overhead from calling CopySendData once per
1163 : : * character, we dump out all characters between escaped characters in a
1164 : : * single call. The loop invariant is that the data from "start" to "ptr"
1165 : : * can be sent literally, but hasn't yet been.
1166 : : *
1167 : : * We can skip pg_encoding_mblen() overhead when encoding is safe, because
1168 : : * in valid backend encodings, extra bytes of a multibyte character never
1169 : : * look like ASCII. This loop is sufficiently performance-critical that
1170 : : * it's worth making two copies of it to get the IS_HIGHBIT_SET() test out
1171 : : * of the normal safe-encoding path.
1172 : : */
1173 [ - + ]: 6489609 : if (cstate->encoding_embeds_ascii)
1174 : : {
1748 heikki.linnakangas@i 1175 :UBC 0 : start = ptr;
1176 [ # # ]: 0 : while ((c = *ptr) != '\0')
1177 : : {
1178 [ # # ]: 0 : if ((unsigned char) c < (unsigned char) 0x20)
1179 : : {
1180 : : /*
1181 : : * \r and \n must be escaped, the others are traditional. We
1182 : : * prefer to dump these using the C-like notation, rather than
1183 : : * a backslash and the literal character, because it makes the
1184 : : * dump file a bit more proof against Microsoftish data
1185 : : * mangling.
1186 : : */
1187 [ # # # # : 0 : switch (c)
# # # ]
1188 : : {
1189 : 0 : case '\b':
1190 : 0 : c = 'b';
1191 : 0 : break;
1192 : 0 : case '\f':
1193 : 0 : c = 'f';
1194 : 0 : break;
1195 : 0 : case '\n':
1196 : 0 : c = 'n';
1197 : 0 : break;
1198 : 0 : case '\r':
1199 : 0 : c = 'r';
1200 : 0 : break;
1201 : 0 : case '\t':
1202 : 0 : c = 't';
1203 : 0 : break;
1204 : 0 : case '\v':
1205 : 0 : c = 'v';
1206 : 0 : break;
1207 : 0 : default:
1208 : : /* If it's the delimiter, must backslash it */
1209 [ # # ]: 0 : if (c == delimc)
1210 : 0 : break;
1211 : : /* All ASCII control chars are length 1 */
1212 : 0 : ptr++;
1213 : 0 : continue; /* fall to end of loop */
1214 : : }
1215 : : /* if we get here, we need to convert the control char */
1216 [ # # ]: 0 : DUMPSOFAR();
1217 : 0 : CopySendChar(cstate, '\\');
1218 : 0 : CopySendChar(cstate, c);
1219 : 0 : start = ++ptr; /* do not include char in next run */
1220 : : }
1221 [ # # # # ]: 0 : else if (c == '\\' || c == delimc)
1222 : : {
1223 [ # # ]: 0 : DUMPSOFAR();
1224 : 0 : CopySendChar(cstate, '\\');
1225 : 0 : start = ptr++; /* we include char in next run */
1226 : : }
1227 [ # # ]: 0 : else if (IS_HIGHBIT_SET(c))
1228 : 0 : ptr += pg_encoding_mblen(cstate->file_encoding, ptr);
1229 : : else
1230 : 0 : ptr++;
1231 : : }
1232 : : }
1233 : : else
1234 : : {
1748 heikki.linnakangas@i 1235 :CBC 6489609 : start = ptr;
1236 [ + + ]: 68499028 : while ((c = *ptr) != '\0')
1237 : : {
1238 [ + + ]: 62009419 : if ((unsigned char) c < (unsigned char) 0x20)
1239 : : {
1240 : : /*
1241 : : * \r and \n must be escaped, the others are traditional. We
1242 : : * prefer to dump these using the C-like notation, rather than
1243 : : * a backslash and the literal character, because it makes the
1244 : : * dump file a bit more proof against Microsoftish data
1245 : : * mangling.
1246 : : */
1247 [ - - + - : 6921 : switch (c)
+ - - ]
1248 : : {
1748 heikki.linnakangas@i 1249 :UBC 0 : case '\b':
1250 : 0 : c = 'b';
1251 : 0 : break;
1252 : 0 : case '\f':
1253 : 0 : c = 'f';
1254 : 0 : break;
1748 heikki.linnakangas@i 1255 :CBC 5864 : case '\n':
1256 : 5864 : c = 'n';
1257 : 5864 : break;
1748 heikki.linnakangas@i 1258 :UBC 0 : case '\r':
1259 : 0 : c = 'r';
1260 : 0 : break;
1748 heikki.linnakangas@i 1261 :CBC 1057 : case '\t':
1262 : 1057 : c = 't';
1263 : 1057 : break;
1748 heikki.linnakangas@i 1264 :UBC 0 : case '\v':
1265 : 0 : c = 'v';
1266 : 0 : break;
1267 : 0 : default:
1268 : : /* If it's the delimiter, must backslash it */
1269 [ # # ]: 0 : if (c == delimc)
1270 : 0 : break;
1271 : : /* All ASCII control chars are length 1 */
1272 : 0 : ptr++;
1273 : 0 : continue; /* fall to end of loop */
1274 : : }
1275 : : /* if we get here, we need to convert the control char */
1748 heikki.linnakangas@i 1276 [ + + ]:CBC 6921 : DUMPSOFAR();
1277 : 6921 : CopySendChar(cstate, '\\');
1278 : 6921 : CopySendChar(cstate, c);
1279 : 6921 : start = ++ptr; /* do not include char in next run */
1280 : : }
1281 [ + + - + ]: 62002498 : else if (c == '\\' || c == delimc)
1282 : : {
1283 [ + + ]: 5812 : DUMPSOFAR();
1284 : 5812 : CopySendChar(cstate, '\\');
1285 : 5812 : start = ptr++; /* we include char in next run */
1286 : : }
1287 : : else
1288 : 61996686 : ptr++;
1289 : : }
1290 : : }
1291 : :
1292 [ + + ]: 6489609 : DUMPSOFAR();
1293 : 6489609 : }
1294 : :
1295 : : /*
1296 : : * Send text representation of one attribute, with conversion and
1297 : : * CSV-style escaping
1298 : : */
1299 : : static void
1317 peter@eisentraut.org 1300 : 309 : CopyAttributeOutCSV(CopyToState cstate, const char *string,
1301 : : bool use_quote)
1302 : : {
1303 : : const char *ptr;
1304 : : const char *start;
1305 : : char c;
1748 heikki.linnakangas@i 1306 : 309 : char delimc = cstate->opts.delim[0];
1307 : 309 : char quotec = cstate->opts.quote[0];
1308 : 309 : char escapec = cstate->opts.escape[0];
577 michael@paquier.xyz 1309 : 309 : bool single_attr = (list_length(cstate->attnumlist) == 1);
1310 : :
1311 : : /* force quoting if it matches null_print (before conversion!) */
1748 heikki.linnakangas@i 1312 [ + + + + ]: 309 : if (!use_quote && strcmp(string, cstate->opts.null_print) == 0)
1313 : 27 : use_quote = true;
1314 : :
1315 [ - + ]: 309 : if (cstate->need_transcoding)
1748 heikki.linnakangas@i 1316 :UBC 0 : ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding);
1317 : : else
1748 heikki.linnakangas@i 1318 :CBC 309 : ptr = string;
1319 : :
1320 : : /*
1321 : : * Make a preliminary pass to discover if it needs quoting
1322 : : */
1323 [ + + ]: 309 : if (!use_quote)
1324 : : {
1325 : : /*
1326 : : * Quote '\.' if it appears alone on a line, so that it will not be
1327 : : * interpreted as an end-of-data marker. (PG 18 and up will not
1328 : : * interpret '\.' in CSV that way, except in embedded-in-SQL data; but
1329 : : * we want the data to be loadable by older versions too. Also, this
1330 : : * avoids breaking clients that are still using PQgetline().)
1331 : : */
1332 [ + + + + ]: 216 : if (single_attr && strcmp(ptr, "\\.") == 0)
1333 : 3 : use_quote = true;
1334 : : else
1335 : : {
1317 peter@eisentraut.org 1336 : 213 : const char *tptr = ptr;
1337 : :
1748 heikki.linnakangas@i 1338 [ + + ]: 1104 : while ((c = *tptr) != '\0')
1339 : : {
1340 [ + + + + : 957 : if (c == delimc || c == quotec || c == '\n' || c == '\r')
+ + + + ]
1341 : : {
1342 : 66 : use_quote = true;
1343 : 66 : break;
1344 : : }
1345 [ + + - + ]: 891 : if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii)
1748 heikki.linnakangas@i 1346 :UBC 0 : tptr += pg_encoding_mblen(cstate->file_encoding, tptr);
1347 : : else
1748 heikki.linnakangas@i 1348 :CBC 891 : tptr++;
1349 : : }
1350 : : }
1351 : : }
1352 : :
1353 [ + + ]: 309 : if (use_quote)
1354 : : {
1355 : 162 : CopySendChar(cstate, quotec);
1356 : :
1357 : : /*
1358 : : * We adopt the same optimization strategy as in CopyAttributeOutText
1359 : : */
1360 : 162 : start = ptr;
1361 [ + + ]: 1269 : while ((c = *ptr) != '\0')
1362 : : {
1363 [ + + + + ]: 1107 : if (c == quotec || c == escapec)
1364 : : {
1365 [ + + ]: 78 : DUMPSOFAR();
1366 : 78 : CopySendChar(cstate, escapec);
1367 : 78 : start = ptr; /* we include char in next run */
1368 : : }
1369 [ - + - - ]: 1107 : if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii)
1748 heikki.linnakangas@i 1370 :UBC 0 : ptr += pg_encoding_mblen(cstate->file_encoding, ptr);
1371 : : else
1748 heikki.linnakangas@i 1372 :CBC 1107 : ptr++;
1373 : : }
1374 [ + + ]: 162 : DUMPSOFAR();
1375 : :
1376 : 162 : CopySendChar(cstate, quotec);
1377 : : }
1378 : : else
1379 : : {
1380 : : /* If it doesn't need quoting, we can just dump it as-is */
1381 : 147 : CopySendString(cstate, ptr);
1382 : : }
1383 : 309 : }
1384 : :
1385 : : /*
1386 : : * copy_dest_startup --- executor startup
1387 : : */
1388 : : static void
1389 : 222 : copy_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
1390 : : {
1391 : : /* no-op */
1392 : 222 : }
1393 : :
1394 : : /*
1395 : : * copy_dest_receive --- receive one tuple
1396 : : */
1397 : : static bool
1398 : 3502 : copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
1399 : : {
1400 : 3502 : DR_copy *myState = (DR_copy *) self;
1578 tgl@sss.pgh.pa.us 1401 : 3502 : CopyToState cstate = myState->cstate;
1402 : :
1403 : : /* Send the data */
1748 heikki.linnakangas@i 1404 : 3502 : CopyOneRowTo(cstate, slot);
1405 : :
1406 : : /* Increment the number of processed tuples, and report the progress */
1642 michael@paquier.xyz 1407 : 3502 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
1408 : 3502 : ++myState->processed);
1409 : :
1748 heikki.linnakangas@i 1410 : 3502 : return true;
1411 : : }
1412 : :
1413 : : /*
1414 : : * copy_dest_shutdown --- executor end
1415 : : */
1416 : : static void
1417 : 222 : copy_dest_shutdown(DestReceiver *self)
1418 : : {
1419 : : /* no-op */
1420 : 222 : }
1421 : :
1422 : : /*
1423 : : * copy_dest_destroy --- release DestReceiver object
1424 : : */
1425 : : static void
1748 heikki.linnakangas@i 1426 :UBC 0 : copy_dest_destroy(DestReceiver *self)
1427 : : {
1428 : 0 : pfree(self);
1429 : 0 : }
1430 : :
1431 : : /*
1432 : : * CreateCopyDestReceiver -- create a suitable DestReceiver object
1433 : : */
1434 : : DestReceiver *
1748 heikki.linnakangas@i 1435 :CBC 225 : CreateCopyDestReceiver(void)
1436 : : {
1437 : 225 : DR_copy *self = (DR_copy *) palloc(sizeof(DR_copy));
1438 : :
1439 : 225 : self->pub.receiveSlot = copy_dest_receive;
1440 : 225 : self->pub.rStartup = copy_dest_startup;
1441 : 225 : self->pub.rShutdown = copy_dest_shutdown;
1442 : 225 : self->pub.rDestroy = copy_dest_destroy;
1443 : 225 : self->pub.mydest = DestCopyOut;
1444 : :
1445 : 225 : self->cstate = NULL; /* will be set later */
1446 : 225 : self->processed = 0;
1447 : :
1448 : 225 : return (DestReceiver *) self;
1449 : : }
|
