Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * copyto.c
4 : : * COPY <table> TO file/program/client
5 : : *
6 : : * Portions Copyright (c) 1996-2026, 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/table.h"
22 : : #include "access/tableam.h"
23 : : #include "access/tupconvert.h"
24 : : #include "access/xlog.h"
25 : : #include "catalog/pg_inherits.h"
26 : : #include "commands/copyapi.h"
27 : : #include "commands/progress.h"
28 : : #include "executor/execdesc.h"
29 : : #include "executor/executor.h"
30 : : #include "executor/tuptable.h"
31 : : #include "funcapi.h"
32 : : #include "libpq/libpq.h"
33 : : #include "libpq/pqformat.h"
34 : : #include "mb/pg_wchar.h"
35 : : #include "miscadmin.h"
36 : : #include "pgstat.h"
37 : : #include "storage/fd.h"
38 : : #include "tcop/tcopprot.h"
39 : : #include "utils/json.h"
40 : : #include "utils/lsyscache.h"
41 : : #include "utils/memutils.h"
42 : : #include "utils/rel.h"
43 : : #include "utils/snapmgr.h"
44 : : #include "utils/wait_event.h"
45 : :
46 : : /*
47 : : * Represents the different dest cases we need to worry about at
48 : : * the bottom level
49 : : */
50 : : typedef enum CopyDest
51 : : {
52 : : COPY_FILE, /* to file (or a piped program) */
53 : : COPY_FRONTEND, /* to frontend */
54 : : COPY_CALLBACK, /* to callback function */
55 : : } CopyDest;
56 : :
57 : : /*
58 : : * This struct contains all the state variables used throughout a COPY TO
59 : : * operation.
60 : : *
61 : : * Multi-byte encodings: all supported client-side encodings encode multi-byte
62 : : * characters by having the first byte's high bit set. Subsequent bytes of the
63 : : * character can have the high bit not set. When scanning data in such an
64 : : * encoding to look for a match to a single-byte (ie ASCII) character, we must
65 : : * use the full pg_encoding_mblen() machinery to skip over multibyte
66 : : * characters, else we might find a false match to a trailing byte. In
67 : : * supported server encodings, there is no possibility of a false match, and
68 : : * it's faster to make useless comparisons to trailing bytes than it is to
69 : : * invoke pg_encoding_mblen() to skip over them. encoding_embeds_ascii is true
70 : : * when we have to do it the hard way.
71 : : */
72 : : typedef struct CopyToStateData
73 : : {
74 : : /* format-specific routines */
75 : : const CopyToRoutine *routine;
76 : :
77 : : /* low-level state data */
78 : : CopyDest copy_dest; /* type of copy source/destination */
79 : : FILE *copy_file; /* used if copy_dest == COPY_FILE */
80 : : StringInfo fe_msgbuf; /* used for all dests during COPY TO */
81 : :
82 : : int file_encoding; /* file or remote side's character encoding */
83 : : bool need_transcoding; /* file encoding diff from server? */
84 : : bool encoding_embeds_ascii; /* ASCII can be non-first byte? */
85 : :
86 : : /* parameters from the COPY command */
87 : : Relation rel; /* relation to copy to */
88 : : QueryDesc *queryDesc; /* executable query to copy from */
89 : : List *attnumlist; /* integer list of attnums to copy */
90 : : char *filename; /* filename, or NULL for STDOUT */
91 : : bool is_program; /* is 'filename' a program to popen? */
92 : : bool json_row_delim_needed; /* need delimiter before next row */
93 : : StringInfo json_buf; /* reusable buffer for JSON output,
94 : : * initialized in BeginCopyTo */
95 : : TupleDesc tupDesc; /* Descriptor for JSON output; for a column
96 : : * list this is a projected descriptor */
97 : : Datum *json_projvalues; /* pre-allocated projection values, or
98 : : * NULL */
99 : : bool *json_projnulls; /* pre-allocated projection nulls, or NULL */
100 : : copy_data_dest_cb data_dest_cb; /* function for writing data */
101 : :
102 : : CopyFormatOptions opts;
103 : : Node *whereClause; /* WHERE condition (or NULL) */
104 : : List *partitions; /* OID list of partitions to copy data from */
105 : :
106 : : /*
107 : : * Working state
108 : : */
109 : : MemoryContext copycontext; /* per-copy execution context */
110 : :
111 : : FmgrInfo *out_functions; /* lookup info for output functions */
112 : : MemoryContext rowcontext; /* per-row evaluation context */
113 : : uint64 bytes_processed; /* number of bytes processed so far */
114 : : } CopyToStateData;
115 : :
116 : : /* DestReceiver for COPY (query) TO */
117 : : typedef struct
118 : : {
119 : : DestReceiver pub; /* publicly-known function pointers */
120 : : CopyToState cstate; /* CopyToStateData for the command */
121 : : uint64 processed; /* # of tuples processed */
122 : : } DR_copy;
123 : :
124 : : /* NOTE: there's a copy of this in copyfromparse.c */
125 : : static const char BinarySignature[11] = "PGCOPY\n\377\r\n\0";
126 : :
127 : :
128 : : /* non-export function prototypes */
129 : : static void EndCopy(CopyToState cstate);
130 : : static void ClosePipeToProgram(CopyToState cstate);
131 : : static void CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot);
132 : : static void CopyAttributeOutText(CopyToState cstate, const char *string);
133 : : static void CopyAttributeOutCSV(CopyToState cstate, const char *string,
134 : : bool use_quote);
135 : : static void CopyRelationTo(CopyToState cstate, Relation rel, Relation root_rel,
136 : : uint64 *processed);
137 : :
138 : : /* built-in format-specific routines */
139 : : static void CopyToTextLikeStart(CopyToState cstate, TupleDesc tupDesc);
140 : : static void CopyToTextLikeOutFunc(CopyToState cstate, Oid atttypid, FmgrInfo *finfo);
141 : : static void CopyToTextOneRow(CopyToState cstate, TupleTableSlot *slot);
142 : : static void CopyToCSVOneRow(CopyToState cstate, TupleTableSlot *slot);
143 : : static void CopyToTextLikeOneRow(CopyToState cstate, TupleTableSlot *slot,
144 : : bool is_csv);
145 : : static void CopyToTextLikeEnd(CopyToState cstate);
146 : : static void CopyToJsonOneRow(CopyToState cstate, TupleTableSlot *slot);
147 : : static void CopyToJsonEnd(CopyToState cstate);
148 : : static void CopyToBinaryStart(CopyToState cstate, TupleDesc tupDesc);
149 : : static void CopyToBinaryOutFunc(CopyToState cstate, Oid atttypid, FmgrInfo *finfo);
150 : : static void CopyToBinaryOneRow(CopyToState cstate, TupleTableSlot *slot);
151 : : static void CopyToBinaryEnd(CopyToState cstate);
152 : :
153 : : /* Low-level communications functions */
154 : : static void SendCopyBegin(CopyToState cstate);
155 : : static void SendCopyEnd(CopyToState cstate);
156 : : static void CopySendData(CopyToState cstate, const void *databuf, int datasize);
157 : : static void CopySendString(CopyToState cstate, const char *str);
158 : : static void CopySendChar(CopyToState cstate, char c);
159 : : static void CopySendEndOfRow(CopyToState cstate);
160 : : static void CopySendTextLikeEndOfRow(CopyToState cstate);
161 : : static void CopySendInt32(CopyToState cstate, int32 val);
162 : : static void CopySendInt16(CopyToState cstate, int16 val);
163 : :
164 : : /*
165 : : * COPY TO routines for built-in formats.
166 : : */
167 : :
168 : : /* text format */
169 : : static const CopyToRoutine CopyToRoutineText = {
170 : : .CopyToStart = CopyToTextLikeStart,
171 : : .CopyToOutFunc = CopyToTextLikeOutFunc,
172 : : .CopyToOneRow = CopyToTextOneRow,
173 : : .CopyToEnd = CopyToTextLikeEnd,
174 : : };
175 : :
176 : : /* CSV format */
177 : : static const CopyToRoutine CopyToRoutineCSV = {
178 : : .CopyToStart = CopyToTextLikeStart,
179 : : .CopyToOutFunc = CopyToTextLikeOutFunc,
180 : : .CopyToOneRow = CopyToCSVOneRow,
181 : : .CopyToEnd = CopyToTextLikeEnd,
182 : : };
183 : :
184 : : /* json format */
185 : : static const CopyToRoutine CopyToRoutineJson = {
186 : : .CopyToStart = CopyToTextLikeStart,
187 : : .CopyToOutFunc = CopyToTextLikeOutFunc,
188 : : .CopyToOneRow = CopyToJsonOneRow,
189 : : .CopyToEnd = CopyToJsonEnd,
190 : : };
191 : :
192 : : /* binary format */
193 : : static const CopyToRoutine CopyToRoutineBinary = {
194 : : .CopyToStart = CopyToBinaryStart,
195 : : .CopyToOutFunc = CopyToBinaryOutFunc,
196 : : .CopyToOneRow = CopyToBinaryOneRow,
197 : : .CopyToEnd = CopyToBinaryEnd,
198 : : };
199 : :
200 : : /* Return a COPY TO routine for the given options */
201 : : static const CopyToRoutine *
202 : 5188 : CopyToGetRoutine(const CopyFormatOptions *opts)
203 : : {
204 [ + + ]: 5188 : if (opts->format == COPY_FORMAT_CSV)
205 : 92 : return &CopyToRoutineCSV;
206 [ + + ]: 5096 : else if (opts->format == COPY_FORMAT_BINARY)
207 : 9 : return &CopyToRoutineBinary;
208 [ + + ]: 5087 : else if (opts->format == COPY_FORMAT_JSON)
209 : 116 : return &CopyToRoutineJson;
210 : :
211 : : /* default is text */
212 : 4971 : return &CopyToRoutineText;
213 : : }
214 : :
215 : : /* Implementation of the start callback for text, CSV, and json formats */
216 : : static void
217 : 5090 : CopyToTextLikeStart(CopyToState cstate, TupleDesc tupDesc)
218 : : {
219 : : /*
220 : : * For non-binary copy, we need to convert null_print to file encoding,
221 : : * because it will be sent directly with CopySendString.
222 : : */
223 [ + + ]: 5090 : if (cstate->need_transcoding)
224 : 17 : cstate->opts.null_print_client = pg_server_to_any(cstate->opts.null_print,
225 : : cstate->opts.null_print_len,
226 : : cstate->file_encoding);
227 : :
228 : : /* if a header has been requested send the line */
229 [ + + ]: 5090 : if (cstate->opts.header_line == COPY_HEADER_TRUE)
230 : : {
231 : : ListCell *cur;
232 : 28 : bool hdr_delim = false;
233 : :
234 : : Assert(cstate->opts.format != COPY_FORMAT_JSON);
235 : :
236 [ + - + + : 76 : foreach(cur, cstate->attnumlist)
+ + ]
237 : : {
238 : 48 : int attnum = lfirst_int(cur);
239 : : char *colname;
240 : :
241 [ + + ]: 48 : if (hdr_delim)
242 : 20 : CopySendChar(cstate, cstate->opts.delim[0]);
243 : 48 : hdr_delim = true;
244 : :
245 : 48 : colname = NameStr(TupleDescAttr(tupDesc, attnum - 1)->attname);
246 : :
247 [ + + ]: 48 : if (cstate->opts.format == COPY_FORMAT_CSV)
248 : 16 : CopyAttributeOutCSV(cstate, colname, false);
249 : : else
250 : 32 : CopyAttributeOutText(cstate, colname);
251 : : }
252 : :
253 : 28 : CopySendTextLikeEndOfRow(cstate);
254 : : }
255 : :
256 : : /*
257 : : * If FORCE_ARRAY has been specified, send the opening bracket.
258 : : */
259 [ + + + + ]: 5090 : if (cstate->opts.format == COPY_FORMAT_JSON && cstate->opts.force_array)
260 : : {
261 : 20 : CopySendChar(cstate, '[');
262 : 20 : CopySendTextLikeEndOfRow(cstate);
263 : : }
264 : 5090 : }
265 : :
266 : : /*
267 : : * Implementation of the outfunc callback for text, CSV, and json formats. Assign
268 : : * the output function data to the given *finfo.
269 : : */
270 : : static void
271 : 16981 : CopyToTextLikeOutFunc(CopyToState cstate, Oid atttypid, FmgrInfo *finfo)
272 : : {
273 : : Oid func_oid;
274 : : bool is_varlena;
275 : :
276 : : /* Set output function for an attribute */
277 : 16981 : getTypeOutputInfo(atttypid, &func_oid, &is_varlena);
278 : 16981 : fmgr_info(func_oid, finfo);
279 : 16981 : }
280 : :
281 : : /* Implementation of the per-row callback for text format */
282 : : static void
283 : 1838910 : CopyToTextOneRow(CopyToState cstate, TupleTableSlot *slot)
284 : : {
285 : 1838910 : CopyToTextLikeOneRow(cstate, slot, false);
286 : 1838910 : }
287 : :
288 : : /* Implementation of the per-row callback for CSV format */
289 : : static void
290 : 228 : CopyToCSVOneRow(CopyToState cstate, TupleTableSlot *slot)
291 : : {
292 : 228 : CopyToTextLikeOneRow(cstate, slot, true);
293 : 228 : }
294 : :
295 : : /*
296 : : * Workhorse for CopyToTextOneRow() and CopyToCSVOneRow().
297 : : *
298 : : * We use pg_always_inline to reduce function call overhead
299 : : * and to help compilers to optimize away the 'is_csv' condition.
300 : : */
301 : : static pg_always_inline void
302 : 1839138 : CopyToTextLikeOneRow(CopyToState cstate,
303 : : TupleTableSlot *slot,
304 : : bool is_csv)
305 : : {
306 : 1839138 : bool need_delim = false;
307 : 1839138 : FmgrInfo *out_functions = cstate->out_functions;
308 : :
309 [ + + + + : 10830943 : foreach_int(attnum, cstate->attnumlist)
+ + ]
310 : : {
311 : 7152667 : Datum value = slot->tts_values[attnum - 1];
312 : 7152667 : bool isnull = slot->tts_isnull[attnum - 1];
313 : :
314 [ + + ]: 7152667 : if (need_delim)
315 : 5313598 : CopySendChar(cstate, cstate->opts.delim[0]);
316 : 7152667 : need_delim = true;
317 : :
318 [ + + ]: 7152667 : if (isnull)
319 : : {
320 : 601753 : CopySendString(cstate, cstate->opts.null_print_client);
321 : : }
322 : : else
323 : : {
324 : : char *string;
325 : :
326 : 6550914 : string = OutputFunctionCall(&out_functions[attnum - 1],
327 : : value);
328 : :
329 [ + + ]: 6550914 : if (is_csv)
330 : 412 : CopyAttributeOutCSV(cstate, string,
331 : 412 : cstate->opts.force_quote_flags[attnum - 1]);
332 : : else
333 : 6550502 : CopyAttributeOutText(cstate, string);
334 : : }
335 : : }
336 : :
337 : 1839138 : CopySendTextLikeEndOfRow(cstate);
338 : 1839138 : }
339 : :
340 : : /* Implementation of the end callback for text and CSV formats */
341 : : static void
342 : 4973 : CopyToTextLikeEnd(CopyToState cstate)
343 : : {
344 : : /* Nothing to do here */
345 : 4973 : }
346 : :
347 : : /* Implementation of the end callback for json format */
348 : : static void
349 : 116 : CopyToJsonEnd(CopyToState cstate)
350 : : {
351 [ + + ]: 116 : if (cstate->opts.force_array)
352 : : {
353 : 20 : CopySendChar(cstate, ']');
354 : 20 : CopySendTextLikeEndOfRow(cstate);
355 : : }
356 : 116 : }
357 : :
358 : : /* Implementation of per-row callback for json format */
359 : : static void
360 : 348 : CopyToJsonOneRow(CopyToState cstate, TupleTableSlot *slot)
361 : : {
362 : : Datum rowdata;
363 : :
364 : 348 : resetStringInfo(cstate->json_buf);
365 : :
366 [ + + ]: 348 : if (cstate->json_projvalues != NULL)
367 : : {
368 : : /*
369 : : * Column list case: project selected column values into sequential
370 : : * positions matching the custom TupleDesc, then form a new tuple.
371 : : */
372 : : HeapTuple tup;
373 : 140 : int i = 0;
374 : :
375 [ + - + + : 560 : foreach_int(attnum, cstate->attnumlist)
+ + ]
376 : : {
377 : 280 : cstate->json_projvalues[i] = slot->tts_values[attnum - 1];
378 : 280 : cstate->json_projnulls[i] = slot->tts_isnull[attnum - 1];
379 : 280 : i++;
380 : : }
381 : :
382 : 140 : tup = heap_form_tuple(cstate->tupDesc,
383 : 140 : cstate->json_projvalues,
384 : 140 : cstate->json_projnulls);
385 : :
386 : : /*
387 : : * heap_form_tuple already stamps the datum-length, type-id, and
388 : : * type-mod fields on t_data, so we can use it directly as a composite
389 : : * Datum without the extra pallocmemcpy that heap_copy_tuple_as_datum
390 : : * would do. Any TOAST pointers in the projected values will be
391 : : * detoasted by the per-column output functions called from
392 : : * composite_to_json.
393 : : */
394 : 140 : rowdata = HeapTupleGetDatum(tup);
395 : : }
396 : : else
397 : : {
398 : : /*
399 : : * Full table or query without column list. For a query, the slot's
400 : : * descriptor is either an unregistered RECORD type, which
401 : : * composite_to_json's lookup_rowtype_tupdesc() cannot look up, or,
402 : : * when the top plan node does not project, the row type of a scanned
403 : : * table, whose column names need not match the query's. Either way,
404 : : * the datum must be stamped with the query's blessed descriptor.
405 : : *
406 : : * A virtual slot has no physical tuple, so form one directly.
407 : : * Otherwise copy the slot's tuple and stamp the copy with the query's
408 : : * descriptor. That is safe because the tuple's physical layout
409 : : * matches the query's result descriptor: a scan returns its scan
410 : : * tuple unprojected only if tlist_matches_tupdesc() holds, which
411 : : * rules out dropped columns and columns with missing values.
412 : : */
413 [ + + ]: 208 : if (cstate->rel)
414 : 128 : rowdata = ExecFetchSlotHeapTupleDatum(slot);
415 [ + + ]: 80 : else if (TTS_IS_VIRTUAL(slot))
416 : 52 : rowdata = HeapTupleGetDatum(heap_form_tuple(cstate->tupDesc,
417 : 52 : slot->tts_values,
418 : 52 : slot->tts_isnull));
419 : : else
420 : : {
421 : : HeapTuple tup;
422 : : bool shouldFree;
423 : :
424 : : Assert(slot->tts_tupleDescriptor->natts == cstate->tupDesc->natts);
425 : :
426 : 28 : tup = ExecFetchSlotHeapTuple(slot, false, &shouldFree);
427 : 28 : rowdata = heap_copy_tuple_as_datum(tup, cstate->tupDesc);
428 [ - + ]: 28 : if (shouldFree)
429 : 0 : heap_freetuple(tup);
430 : : }
431 : : }
432 : :
433 : 348 : composite_to_json(rowdata, cstate->json_buf, false);
434 : :
435 [ + + ]: 348 : if (cstate->opts.force_array)
436 : : {
437 [ + + ]: 48 : if (cstate->json_row_delim_needed)
438 : 32 : CopySendChar(cstate, ',');
439 : : else
440 : : {
441 : : /* first row needs no delimiter */
442 : 16 : CopySendChar(cstate, ' ');
443 : 16 : cstate->json_row_delim_needed = true;
444 : : }
445 : : }
446 : :
447 : : /*
448 : : * Convert the JSON output to the target encoding if needed. Unlike the
449 : : * text and CSV paths which convert per-attribute via CopyAttributeOut*,
450 : : * composite_to_json() emits the whole row as one buffer, so we transcode
451 : : * it here in a single call before sending.
452 : : */
453 [ + + ]: 348 : if (cstate->need_transcoding)
454 : : {
455 : : char *converted;
456 : :
457 : 8 : converted = pg_server_to_any(cstate->json_buf->data,
458 : 8 : cstate->json_buf->len,
459 : : cstate->file_encoding);
460 : 8 : CopySendData(cstate, converted, strlen(converted));
461 [ + - ]: 8 : if (converted != cstate->json_buf->data)
462 : 8 : pfree(converted);
463 : : }
464 : : else
465 : 340 : CopySendData(cstate, cstate->json_buf->data, cstate->json_buf->len);
466 : :
467 : 348 : CopySendTextLikeEndOfRow(cstate);
468 : 348 : }
469 : :
470 : : /*
471 : : * Implementation of the start callback for binary format. Send a header
472 : : * for a binary copy.
473 : : */
474 : : static void
475 : 8 : CopyToBinaryStart(CopyToState cstate, TupleDesc tupDesc)
476 : : {
477 : : int32 tmp;
478 : :
479 : : /* Signature */
480 : 8 : CopySendData(cstate, BinarySignature, 11);
481 : : /* Flags field */
482 : 8 : tmp = 0;
483 : 8 : CopySendInt32(cstate, tmp);
484 : : /* No header extension */
485 : 8 : tmp = 0;
486 : 8 : CopySendInt32(cstate, tmp);
487 : 8 : }
488 : :
489 : : /*
490 : : * Implementation of the outfunc callback for binary format. Assign
491 : : * the binary output function to the given *finfo.
492 : : */
493 : : static void
494 : 38 : CopyToBinaryOutFunc(CopyToState cstate, Oid atttypid, FmgrInfo *finfo)
495 : : {
496 : : Oid func_oid;
497 : : bool is_varlena;
498 : :
499 : : /* Set output function for an attribute */
500 : 38 : getTypeBinaryOutputInfo(atttypid, &func_oid, &is_varlena);
501 : 37 : fmgr_info(func_oid, finfo);
502 : 37 : }
503 : :
504 : : /* Implementation of the per-row callback for binary format */
505 : : static void
506 : 19 : CopyToBinaryOneRow(CopyToState cstate, TupleTableSlot *slot)
507 : : {
508 : 19 : FmgrInfo *out_functions = cstate->out_functions;
509 : :
510 : : /* Binary per-tuple header */
511 : 19 : CopySendInt16(cstate, list_length(cstate->attnumlist));
512 : :
513 [ + - + + : 139 : foreach_int(attnum, cstate->attnumlist)
+ + ]
514 : : {
515 : 101 : Datum value = slot->tts_values[attnum - 1];
516 : 101 : bool isnull = slot->tts_isnull[attnum - 1];
517 : :
518 [ + + ]: 101 : if (isnull)
519 : : {
520 : 20 : CopySendInt32(cstate, -1);
521 : : }
522 : : else
523 : : {
524 : : bytea *outputbytes;
525 : :
526 : 81 : outputbytes = SendFunctionCall(&out_functions[attnum - 1],
527 : : value);
528 : 81 : CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
529 : 81 : CopySendData(cstate, VARDATA(outputbytes),
530 : 81 : VARSIZE(outputbytes) - VARHDRSZ);
531 : : }
532 : : }
533 : :
534 : 19 : CopySendEndOfRow(cstate);
535 : 19 : }
536 : :
537 : : /* Implementation of the end callback for binary format */
538 : : static void
539 : 8 : CopyToBinaryEnd(CopyToState cstate)
540 : : {
541 : : /* Generate trailer for a binary copy */
542 : 8 : CopySendInt16(cstate, -1);
543 : : /* Need to flush out the trailer */
544 : 8 : CopySendEndOfRow(cstate);
545 : 8 : }
546 : :
547 : : /*
548 : : * Send copy start/stop messages for frontend copies. These have changed
549 : : * in past protocol redesigns.
550 : : */
551 : : static void
552 : 5041 : SendCopyBegin(CopyToState cstate)
553 : : {
554 : : StringInfoData buf;
555 : 5041 : int natts = list_length(cstate->attnumlist);
556 : 5041 : int16 format = (cstate->opts.format == COPY_FORMAT_BINARY ? 1 : 0);
557 : : int i;
558 : :
559 : 5041 : pq_beginmessage(&buf, PqMsg_CopyOutResponse);
560 : 5041 : pq_sendbyte(&buf, format); /* overall format */
561 [ + + ]: 5041 : if (cstate->opts.format != COPY_FORMAT_JSON)
562 : : {
563 : 4933 : pq_sendint16(&buf, natts);
564 [ + + ]: 21523 : for (i = 0; i < natts; i++)
565 : 16590 : pq_sendint16(&buf, format); /* per-column formats */
566 : : }
567 : : else
568 : : {
569 : : /*
570 : : * For JSON format, report one text-format column. Each CopyData
571 : : * message contains one complete JSON object, not individual column
572 : : * values, so the per-column count is always 1.
573 : : */
574 : 108 : pq_sendint16(&buf, 1);
575 : 108 : pq_sendint16(&buf, 0);
576 : : }
577 : :
578 : 5041 : pq_endmessage(&buf);
579 : 5041 : cstate->copy_dest = COPY_FRONTEND;
580 : 5041 : }
581 : :
582 : : static void
583 : 5039 : SendCopyEnd(CopyToState cstate)
584 : : {
585 : : /* Shouldn't have any unsent data */
586 : : Assert(cstate->fe_msgbuf->len == 0);
587 : : /* Send Copy Done message */
588 : 5039 : pq_putemptymessage(PqMsg_CopyDone);
589 : 5039 : }
590 : :
591 : : /*----------
592 : : * CopySendData sends output data to the destination (file or frontend)
593 : : * CopySendString does the same for null-terminated strings
594 : : * CopySendChar does the same for single characters
595 : : * CopySendEndOfRow does the appropriate thing at end of each data row
596 : : * (data is not actually flushed except by CopySendEndOfRow)
597 : : *
598 : : * NB: no data conversion is applied by these functions
599 : : *----------
600 : : */
601 : : static void
602 : 6449119 : CopySendData(CopyToState cstate, const void *databuf, int datasize)
603 : : {
604 : 6449119 : appendBinaryStringInfo(cstate->fe_msgbuf, databuf, datasize);
605 : 6449119 : }
606 : :
607 : : static void
608 : 601961 : CopySendString(CopyToState cstate, const char *str)
609 : : {
610 : 601961 : appendBinaryStringInfo(cstate->fe_msgbuf, str, strlen(str));
611 : 601961 : }
612 : :
613 : : static void
614 : 7170104 : CopySendChar(CopyToState cstate, char c)
615 : : {
616 [ + + ]: 7170104 : appendStringInfoCharMacro(cstate->fe_msgbuf, c);
617 : 7170104 : }
618 : :
619 : : static void
620 : 1839581 : CopySendEndOfRow(CopyToState cstate)
621 : : {
622 : 1839581 : StringInfo fe_msgbuf = cstate->fe_msgbuf;
623 : :
624 [ + + + - ]: 1839581 : switch (cstate->copy_dest)
625 : : {
626 : 8203 : case COPY_FILE:
627 : 8203 : pgstat_report_wait_start(WAIT_EVENT_COPY_TO_WRITE);
628 [ + - ]: 8203 : if (fwrite(fe_msgbuf->data, fe_msgbuf->len, 1,
629 [ - + ]: 8203 : cstate->copy_file) != 1 ||
630 : 8203 : ferror(cstate->copy_file))
631 : : {
632 [ # # ]: 0 : if (cstate->is_program)
633 : : {
634 [ # # ]: 0 : if (errno == EPIPE)
635 : : {
636 : : /*
637 : : * The pipe will be closed automatically on error at
638 : : * the end of transaction, but we might get a better
639 : : * error message from the subprocess' exit code than
640 : : * just "Broken Pipe"
641 : : */
642 : 0 : ClosePipeToProgram(cstate);
643 : :
644 : : /*
645 : : * If ClosePipeToProgram() didn't throw an error, the
646 : : * program terminated normally, but closed the pipe
647 : : * first. Restore errno, and throw an error.
648 : : */
649 : 0 : errno = EPIPE;
650 : : }
651 [ # # ]: 0 : ereport(ERROR,
652 : : (errcode_for_file_access(),
653 : : errmsg("could not write to COPY program: %m")));
654 : : }
655 : : else
656 [ # # ]: 0 : ereport(ERROR,
657 : : (errcode_for_file_access(),
658 : : errmsg("could not write to COPY file: %m")));
659 : : }
660 : 8203 : pgstat_report_wait_end();
661 : 8203 : break;
662 : 1831375 : case COPY_FRONTEND:
663 : : /* Dump the accumulated row as one CopyData message */
664 : 1831375 : (void) pq_putmessage(PqMsg_CopyData, fe_msgbuf->data, fe_msgbuf->len);
665 : 1831375 : break;
666 : 3 : case COPY_CALLBACK:
667 : 3 : cstate->data_dest_cb(fe_msgbuf->data, fe_msgbuf->len);
668 : 3 : break;
669 : : }
670 : :
671 : : /* Update the progress */
672 : 1839581 : cstate->bytes_processed += fe_msgbuf->len;
673 : 1839581 : pgstat_progress_update_param(PROGRESS_COPY_BYTES_PROCESSED, cstate->bytes_processed);
674 : :
675 : 1839581 : resetStringInfo(fe_msgbuf);
676 : 1839581 : }
677 : :
678 : : /*
679 : : * Wrapper function of CopySendEndOfRow for text, CSV, and json formats. Sends the
680 : : * line termination and do common appropriate things for the end of row.
681 : : */
682 : : static inline void
683 : 1839554 : CopySendTextLikeEndOfRow(CopyToState cstate)
684 : : {
685 [ + + + ]: 1839554 : switch (cstate->copy_dest)
686 : : {
687 : 8187 : case COPY_FILE:
688 : : /* Default line termination depends on platform */
689 : : #ifndef WIN32
690 : 8187 : CopySendChar(cstate, '\n');
691 : : #else
692 : : CopySendString(cstate, "\r\n");
693 : : #endif
694 : 8187 : break;
695 : 1831364 : case COPY_FRONTEND:
696 : : /* The FE/BE protocol uses \n as newline for all platforms */
697 : 1831364 : CopySendChar(cstate, '\n');
698 : 1831364 : break;
699 : 3 : default:
700 : 3 : break;
701 : : }
702 : :
703 : : /* Now take the actions related to the end of a row */
704 : 1839554 : CopySendEndOfRow(cstate);
705 : 1839554 : }
706 : :
707 : : /*
708 : : * These functions do apply some data conversion
709 : : */
710 : :
711 : : /*
712 : : * CopySendInt32 sends an int32 in network byte order
713 : : */
714 : : static inline void
715 : 117 : CopySendInt32(CopyToState cstate, int32 val)
716 : : {
717 : : uint32 buf;
718 : :
719 : 117 : buf = pg_hton32((uint32) val);
720 : 117 : CopySendData(cstate, &buf, sizeof(buf));
721 : 117 : }
722 : :
723 : : /*
724 : : * CopySendInt16 sends an int16 in network byte order
725 : : */
726 : : static inline void
727 : 27 : CopySendInt16(CopyToState cstate, int16 val)
728 : : {
729 : : uint16 buf;
730 : :
731 : 27 : buf = pg_hton16((uint16) val);
732 : 27 : CopySendData(cstate, &buf, sizeof(buf));
733 : 27 : }
734 : :
735 : : /*
736 : : * Closes the pipe to an external program, checking the pclose() return code.
737 : : */
738 : : static void
739 : 0 : ClosePipeToProgram(CopyToState cstate)
740 : : {
741 : : int pclose_rc;
742 : :
743 : : Assert(cstate->is_program);
744 : :
745 : 0 : pclose_rc = ClosePipeStream(cstate->copy_file);
746 [ # # ]: 0 : if (pclose_rc == -1)
747 [ # # ]: 0 : ereport(ERROR,
748 : : (errcode_for_file_access(),
749 : : errmsg("could not close pipe to external command: %m")));
750 [ # # ]: 0 : else if (pclose_rc != 0)
751 : : {
752 [ # # ]: 0 : ereport(ERROR,
753 : : (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
754 : : errmsg("program \"%s\" failed",
755 : : cstate->filename),
756 : : errdetail_internal("%s", wait_result_to_str(pclose_rc))));
757 : : }
758 : 0 : }
759 : :
760 : : /*
761 : : * Release resources allocated in a cstate for COPY TO.
762 : : */
763 : : static void
764 : 5097 : EndCopy(CopyToState cstate)
765 : : {
766 [ - + ]: 5097 : if (cstate->is_program)
767 : : {
768 : 0 : ClosePipeToProgram(cstate);
769 : : }
770 : : else
771 : : {
772 [ + + - + ]: 5097 : if (cstate->filename != NULL && FreeFile(cstate->copy_file))
773 [ # # ]: 0 : ereport(ERROR,
774 : : (errcode_for_file_access(),
775 : : errmsg("could not close file \"%s\": %m",
776 : : cstate->filename)));
777 : : }
778 : :
779 : 5097 : pgstat_progress_end_command();
780 : :
781 : 5097 : MemoryContextDelete(cstate->copycontext);
782 : :
783 [ + + ]: 5097 : if (cstate->partitions)
784 : 24 : list_free(cstate->partitions);
785 : :
786 : 5097 : pfree(cstate);
787 : 5097 : }
788 : :
789 : : /*
790 : : * Setup CopyToState to read tuples from a table or a query for COPY TO.
791 : : *
792 : : * 'rel': Relation to be copied
793 : : * 'raw_query': Query whose results are to be copied
794 : : * 'queryRelId': OID of base relation to convert to a query (for RLS)
795 : : * 'filename': Name of server-local file to write, NULL for STDOUT
796 : : * 'is_program': true if 'filename' is program to execute
797 : : * 'data_dest_cb': Callback that processes the output data
798 : : * 'attnamelist': List of char *, columns to include. NIL selects all cols.
799 : : * 'options': List of DefElem. See copy_opt_item in gram.y for selections.
800 : : *
801 : : * Returns a CopyToState, to be passed to DoCopyTo() and related functions.
802 : : */
803 : : CopyToState
804 : 5299 : BeginCopyTo(ParseState *pstate,
805 : : Relation rel,
806 : : RawStmt *raw_query,
807 : : Oid queryRelId,
808 : : const char *filename,
809 : : bool is_program,
810 : : copy_data_dest_cb data_dest_cb,
811 : : List *attnamelist,
812 : : List *options)
813 : : {
814 : : CopyToState cstate;
815 [ + + + + ]: 5299 : bool pipe = (filename == NULL && data_dest_cb == NULL);
816 : : TupleDesc tupDesc;
817 : : int num_phys_attrs;
818 : : MemoryContext oldcontext;
819 : 5299 : const int progress_cols[] = {
820 : : PROGRESS_COPY_COMMAND,
821 : : PROGRESS_COPY_TYPE
822 : : };
823 : 5299 : int64 progress_vals[] = {
824 : : PROGRESS_COPY_COMMAND_TO,
825 : : 0
826 : : };
827 : 5299 : List *children = NIL;
828 : :
829 [ + + + + ]: 5299 : if (rel != NULL && rel->rd_rel->relkind != RELKIND_RELATION)
830 : : {
831 [ + + ]: 42 : if (rel->rd_rel->relkind == RELKIND_VIEW)
832 [ + - ]: 8 : ereport(ERROR,
833 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
834 : : errmsg("cannot copy from view \"%s\"",
835 : : RelationGetRelationName(rel)),
836 : : errhint("Try the COPY (SELECT ...) TO variant.")));
837 [ + + ]: 34 : else if (rel->rd_rel->relkind == RELKIND_MATVIEW)
838 : : {
839 [ + + ]: 8 : if (!RelationIsPopulated(rel))
840 [ + - ]: 4 : ereport(ERROR,
841 : : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
842 : : errmsg("cannot copy from unpopulated materialized view \"%s\"",
843 : : RelationGetRelationName(rel)),
844 : : errhint("Use the REFRESH MATERIALIZED VIEW command."));
845 : : }
846 [ - + ]: 26 : else if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
847 [ # # ]: 0 : ereport(ERROR,
848 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
849 : : errmsg("cannot copy from foreign table \"%s\"",
850 : : RelationGetRelationName(rel)),
851 : : errhint("Try the COPY (SELECT ...) TO variant.")));
852 [ - + ]: 26 : else if (rel->rd_rel->relkind == RELKIND_SEQUENCE)
853 [ # # ]: 0 : ereport(ERROR,
854 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
855 : : errmsg("cannot copy from sequence \"%s\"",
856 : : RelationGetRelationName(rel))));
857 [ + - ]: 26 : else if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
858 : : {
859 : : /*
860 : : * Collect OIDs of relation containing data, so that later
861 : : * DoCopyTo can copy the data from them.
862 : : */
863 : 26 : children = find_all_inheritors(RelationGetRelid(rel), AccessShareLock, NULL);
864 : :
865 [ + - + + : 126 : foreach_oid(child, children)
+ + ]
866 : : {
867 : 78 : char relkind = get_rel_relkind(child);
868 : :
869 [ + + ]: 78 : if (relkind == RELKIND_FOREIGN_TABLE)
870 : : {
871 : 1 : char *relation_name = get_rel_name(child);
872 : :
873 [ + - ]: 1 : ereport(ERROR,
874 : : errcode(ERRCODE_WRONG_OBJECT_TYPE),
875 : : errmsg("cannot copy from foreign table \"%s\"", relation_name),
876 : : errdetail("Partition \"%s\" is a foreign table in partitioned table \"%s\".",
877 : : relation_name, RelationGetRelationName(rel)),
878 : : errhint("Try the COPY (SELECT ...) TO variant."));
879 : : }
880 : :
881 : : /* Unlogged partitions are inaccessible during recovery. */
882 [ + + ]: 77 : if (RecoveryInProgress() &&
883 [ + + ]: 4 : get_rel_persistence(child) == RELPERSISTENCE_UNLOGGED)
884 : : {
885 : 1 : char *relation_name = get_rel_name(child);
886 : :
887 [ + - ]: 1 : ereport(ERROR,
888 : : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
889 : : errmsg("cannot access temporary or unlogged relations during recovery"),
890 : : errdetail("Partition \"%s\" is an unlogged table in partitioned table \"%s\".",
891 : : relation_name, RelationGetRelationName(rel)));
892 : : }
893 : :
894 : : /* Exclude tables with no data */
895 [ + + - + ]: 76 : if (RELKIND_HAS_PARTITIONS(relkind))
896 : 37 : children = foreach_delete_current(children, child);
897 : : }
898 : : }
899 : : else
900 [ # # ]: 0 : ereport(ERROR,
901 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
902 : : errmsg("cannot copy from non-table relation \"%s\"",
903 : : RelationGetRelationName(rel))));
904 : : }
905 : :
906 : : /* Temporary and unlogged relations are inaccessible during recovery. */
907 [ + + + + : 5285 : if (rel != NULL && !RelationIsPermanent(rel) && RecoveryInProgress())
+ + ]
908 [ + - ]: 1 : ereport(ERROR,
909 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
910 : : errmsg("cannot access temporary or unlogged relations during recovery")));
911 : :
912 : : /* Allocate workspace and zero all fields */
913 : 5284 : cstate = palloc0_object(CopyToStateData);
914 : :
915 : : /*
916 : : * We allocate everything used by a cstate in a new memory context. This
917 : : * avoids memory leaks during repeated use of COPY in a query.
918 : : */
919 : 5284 : cstate->copycontext = AllocSetContextCreate(CurrentMemoryContext,
920 : : "COPY",
921 : : ALLOCSET_DEFAULT_SIZES);
922 : :
923 : 5284 : oldcontext = MemoryContextSwitchTo(cstate->copycontext);
924 : :
925 : : /* Extract options from the statement node tree */
926 : 5284 : ProcessCopyOptions(pstate, &cstate->opts, false /* is_from */ , options);
927 : :
928 : : /* Set format routine */
929 : 5188 : cstate->routine = CopyToGetRoutine(&cstate->opts);
930 : :
931 : : /* Process the source/target relation or query */
932 [ + + ]: 5188 : if (rel)
933 : : {
934 : : Assert(!raw_query);
935 : :
936 : 4767 : cstate->rel = rel;
937 : :
938 : 4767 : tupDesc = RelationGetDescr(cstate->rel);
939 : 4767 : cstate->partitions = children;
940 : 4767 : cstate->tupDesc = tupDesc;
941 : : }
942 : : else
943 : : {
944 : : List *rewritten;
945 : : Query *query;
946 : : PlannedStmt *plan;
947 : : DestReceiver *dest;
948 : :
949 : 421 : cstate->rel = NULL;
950 : 421 : cstate->partitions = NIL;
951 : :
952 : : /*
953 : : * Run parse analysis and rewrite. Note this also acquires sufficient
954 : : * locks on the source table(s).
955 : : */
956 : 421 : rewritten = pg_analyze_and_rewrite_fixedparams(raw_query,
957 : : pstate->p_sourcetext, NULL, 0,
958 : : NULL);
959 : :
960 : : /* check that we got back something we can work with */
961 [ + + ]: 413 : if (rewritten == NIL)
962 : : {
963 [ + - ]: 12 : ereport(ERROR,
964 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
965 : : errmsg("DO INSTEAD NOTHING rules are not supported for COPY")));
966 : : }
967 [ + + ]: 401 : else if (list_length(rewritten) > 1)
968 : : {
969 : : ListCell *lc;
970 : :
971 : : /* examine queries to determine which error message to issue */
972 [ + - + + : 68 : foreach(lc, rewritten)
+ + ]
973 : : {
974 : 56 : Query *q = lfirst_node(Query, lc);
975 : :
976 [ + + ]: 56 : if (q->querySource == QSRC_QUAL_INSTEAD_RULE)
977 [ + - ]: 12 : ereport(ERROR,
978 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
979 : : errmsg("conditional DO INSTEAD rules are not supported for COPY")));
980 [ + + ]: 44 : if (q->querySource == QSRC_NON_INSTEAD_RULE)
981 [ + - ]: 12 : ereport(ERROR,
982 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
983 : : errmsg("DO ALSO rules are not supported for COPY")));
984 : : }
985 : :
986 [ + - ]: 12 : ereport(ERROR,
987 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
988 : : errmsg("multi-statement DO INSTEAD rules are not supported for COPY")));
989 : : }
990 : :
991 : 365 : query = linitial_node(Query, rewritten);
992 : :
993 : : /* The grammar allows SELECT INTO, but we don't support that */
994 [ + + ]: 365 : if (query->utilityStmt != NULL &&
995 [ + + ]: 12 : IsA(query->utilityStmt, CreateTableAsStmt))
996 [ + - ]: 8 : ereport(ERROR,
997 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
998 : : errmsg("COPY (SELECT INTO) is not supported")));
999 : :
1000 : : /* The only other utility command we could see is NOTIFY */
1001 [ + + ]: 357 : if (query->utilityStmt != NULL)
1002 [ + - ]: 4 : ereport(ERROR,
1003 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1004 : : errmsg("COPY query must not be a utility command")));
1005 : :
1006 : : /*
1007 : : * Similarly the grammar doesn't enforce the presence of a RETURNING
1008 : : * clause, but this is required here.
1009 : : */
1010 [ + + ]: 353 : if (query->commandType != CMD_SELECT &&
1011 [ + + ]: 69 : query->returningList == NIL)
1012 : : {
1013 : : Assert(query->commandType == CMD_INSERT ||
1014 : : query->commandType == CMD_UPDATE ||
1015 : : query->commandType == CMD_DELETE ||
1016 : : query->commandType == CMD_MERGE);
1017 : :
1018 [ + - ]: 16 : ereport(ERROR,
1019 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1020 : : errmsg("COPY query must have a RETURNING clause")));
1021 : : }
1022 : :
1023 : : /* plan the query */
1024 : 337 : plan = pg_plan_query(query, pstate->p_sourcetext,
1025 : : CURSOR_OPT_PARALLEL_OK, NULL, NULL);
1026 : :
1027 : : /*
1028 : : * With row-level security and a user using "COPY relation TO", we
1029 : : * have to convert the "COPY relation TO" to a query-based COPY (eg:
1030 : : * "COPY (SELECT * FROM ONLY relation) TO"), to allow the rewriter to
1031 : : * add in any RLS clauses.
1032 : : *
1033 : : * When this happens, we are passed in the relid of the originally
1034 : : * found relation (which we have locked). As the planner will look up
1035 : : * the relation again, we double-check here to make sure it found the
1036 : : * same one that we have locked.
1037 : : */
1038 [ + + ]: 336 : if (queryRelId != InvalidOid)
1039 : : {
1040 : : /*
1041 : : * Note that with RLS involved there may be multiple relations,
1042 : : * and while the one we need is almost certainly first, we don't
1043 : : * make any guarantees of that in the planner, so check the whole
1044 : : * list and make sure we find the original relation.
1045 : : */
1046 [ - + ]: 52 : if (!list_member_oid(plan->relationOids, queryRelId))
1047 [ # # ]: 0 : ereport(ERROR,
1048 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1049 : : errmsg("relation referenced by COPY statement has changed")));
1050 : : }
1051 : :
1052 : : /*
1053 : : * Use a snapshot with an updated command ID to ensure this query sees
1054 : : * results of any previously executed queries.
1055 : : */
1056 : 336 : PushCopiedSnapshot(GetActiveSnapshot());
1057 : 336 : UpdateActiveSnapshotCommandId();
1058 : :
1059 : : /* Create dest receiver for COPY OUT */
1060 : 336 : dest = CreateDestReceiver(DestCopyOut);
1061 : 336 : ((DR_copy *) dest)->cstate = cstate;
1062 : :
1063 : : /* Create a QueryDesc requesting no output */
1064 : 336 : cstate->queryDesc = CreateQueryDesc(plan, pstate->p_sourcetext,
1065 : : GetActiveSnapshot(),
1066 : : InvalidSnapshot,
1067 : : dest, NULL, NULL, 0);
1068 : :
1069 : : /*
1070 : : * Call ExecutorStart to prepare the plan for execution.
1071 : : *
1072 : : * ExecutorStart computes a result tupdesc for us
1073 : : */
1074 : 336 : ExecutorStart(cstate->queryDesc, 0);
1075 : :
1076 : 332 : tupDesc = cstate->queryDesc->tupDesc;
1077 : 332 : tupDesc = BlessTupleDesc(tupDesc);
1078 : 332 : cstate->tupDesc = tupDesc;
1079 : : }
1080 : :
1081 : : /* Generate or convert list of attributes to process */
1082 : 5099 : cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
1083 : :
1084 : : /* Set up JSON-specific state */
1085 [ + + ]: 5099 : if (cstate->opts.format == COPY_FORMAT_JSON)
1086 : : {
1087 : 116 : cstate->json_buf = makeStringInfo();
1088 : :
1089 : : /*
1090 : : * Build a projected TupleDesc describing only the selected columns so
1091 : : * that composite_to_json() emits the right column names and types;
1092 : : * needed when an explicit column list was given (possibly with a
1093 : : * different column order) or when generated columns are excluded from
1094 : : * the output.
1095 : : */
1096 [ + + + + ]: 116 : if (rel && (attnamelist != NIL ||
1097 [ + + ]: 32 : list_length(cstate->attnumlist) < tupDesc->natts))
1098 : : {
1099 : 44 : int natts = list_length(cstate->attnumlist);
1100 : : TupleDesc resultDesc;
1101 : :
1102 : 44 : resultDesc = CreateTemplateTupleDesc(natts);
1103 : :
1104 [ + - + + : 180 : foreach_int(attnum, cstate->attnumlist)
+ + ]
1105 : : {
1106 : 92 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1107 : :
1108 : 92 : TupleDescInitEntry(resultDesc,
1109 : 92 : foreach_current_index(attnum) + 1,
1110 : 92 : NameStr(attr->attname),
1111 : : attr->atttypid,
1112 : : attr->atttypmod,
1113 : 92 : attr->attndims);
1114 : : }
1115 : :
1116 : 44 : TupleDescFinalize(resultDesc);
1117 : 44 : cstate->tupDesc = BlessTupleDesc(resultDesc);
1118 : :
1119 : : /*
1120 : : * Pre-allocate arrays for projecting selected column values into
1121 : : * sequential positions matching the custom TupleDesc.
1122 : : */
1123 : 44 : cstate->json_projvalues = palloc_array(Datum, natts);
1124 : 44 : cstate->json_projnulls = palloc_array(bool, natts);
1125 : : }
1126 : : }
1127 : :
1128 : 5099 : num_phys_attrs = tupDesc->natts;
1129 : :
1130 : : /* Convert FORCE_QUOTE name list to per-column flags, check validity */
1131 : 5099 : cstate->opts.force_quote_flags = palloc0_array(bool, num_phys_attrs);
1132 [ + + ]: 5099 : if (cstate->opts.force_quote_all)
1133 : : {
1134 [ + - - + : 12 : MemSet(cstate->opts.force_quote_flags, true, num_phys_attrs * sizeof(bool));
- - - - -
- ]
1135 : : }
1136 [ + + ]: 5087 : else if (cstate->opts.force_quote)
1137 : : {
1138 : : List *attnums;
1139 : : ListCell *cur;
1140 : :
1141 : 16 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_quote);
1142 : :
1143 [ + - + + : 32 : foreach(cur, attnums)
+ + ]
1144 : : {
1145 : 16 : int attnum = lfirst_int(cur);
1146 : 16 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1147 : :
1148 [ - + ]: 16 : if (!list_member_int(cstate->attnumlist, attnum))
1149 [ # # ]: 0 : ereport(ERROR,
1150 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1151 : : /*- translator: %s is the name of a COPY option, e.g. FORCE_NOT_NULL */
1152 : : errmsg("%s column \"%s\" not referenced by COPY",
1153 : : "FORCE_QUOTE", NameStr(attr->attname))));
1154 : 16 : cstate->opts.force_quote_flags[attnum - 1] = true;
1155 : : }
1156 : : }
1157 : :
1158 : : /* Use client encoding when ENCODING option is not specified. */
1159 [ + + ]: 5099 : if (cstate->opts.file_encoding < 0)
1160 : 5075 : cstate->file_encoding = pg_get_client_encoding();
1161 : : else
1162 : 24 : cstate->file_encoding = cstate->opts.file_encoding;
1163 : :
1164 : : /*
1165 : : * Set up encoding conversion info if the file and server encodings differ
1166 : : * (see also pg_server_to_any).
1167 : : */
1168 [ + + ]: 5099 : if (cstate->file_encoding == GetDatabaseEncoding() ||
1169 [ + + ]: 21 : cstate->file_encoding == PG_SQL_ASCII)
1170 : 5082 : cstate->need_transcoding = false;
1171 : : else
1172 : 17 : cstate->need_transcoding = true;
1173 : :
1174 : : /* See Multibyte encoding comment above */
1175 [ + + + - ]: 5099 : cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding);
1176 : :
1177 : 5099 : cstate->copy_dest = COPY_FILE; /* default */
1178 : :
1179 [ + + ]: 5099 : if (data_dest_cb)
1180 : : {
1181 : 1 : progress_vals[1] = PROGRESS_COPY_TYPE_CALLBACK;
1182 : 1 : cstate->copy_dest = COPY_CALLBACK;
1183 : 1 : cstate->data_dest_cb = data_dest_cb;
1184 : : }
1185 [ + + ]: 5098 : else if (pipe)
1186 : : {
1187 : 5041 : progress_vals[1] = PROGRESS_COPY_TYPE_PIPE;
1188 : :
1189 : : Assert(!is_program); /* the grammar does not allow this */
1190 [ - + ]: 5041 : if (whereToSendOutput != DestRemote)
1191 : 0 : cstate->copy_file = stdout;
1192 : : }
1193 : : else
1194 : : {
1195 : 57 : cstate->filename = pstrdup(filename);
1196 : 57 : cstate->is_program = is_program;
1197 : :
1198 [ - + ]: 57 : if (is_program)
1199 : : {
1200 : 0 : progress_vals[1] = PROGRESS_COPY_TYPE_PROGRAM;
1201 : 0 : cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_W);
1202 [ # # ]: 0 : if (cstate->copy_file == NULL)
1203 [ # # ]: 0 : ereport(ERROR,
1204 : : (errcode_for_file_access(),
1205 : : errmsg("could not execute command \"%s\": %m",
1206 : : cstate->filename)));
1207 : : }
1208 : : else
1209 : : {
1210 : : mode_t oumask; /* Pre-existing umask value */
1211 : : struct stat st;
1212 : :
1213 : 57 : progress_vals[1] = PROGRESS_COPY_TYPE_FILE;
1214 : :
1215 : : /*
1216 : : * Prevent write to relative path ... too easy to shoot oneself in
1217 : : * the foot by overwriting a database file ...
1218 : : */
1219 [ - + ]: 57 : if (!is_absolute_path(filename))
1220 [ # # ]: 0 : ereport(ERROR,
1221 : : (errcode(ERRCODE_INVALID_NAME),
1222 : : errmsg("relative path not allowed for COPY to file")));
1223 : :
1224 : 57 : oumask = umask(S_IWGRP | S_IWOTH);
1225 [ + - ]: 57 : PG_TRY();
1226 : : {
1227 : 57 : cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_W);
1228 : : }
1229 : 0 : PG_FINALLY();
1230 : : {
1231 : 57 : umask(oumask);
1232 : : }
1233 [ - + ]: 57 : PG_END_TRY();
1234 [ - + ]: 57 : if (cstate->copy_file == NULL)
1235 : : {
1236 : : /* copy errno because ereport subfunctions might change it */
1237 : 0 : int save_errno = errno;
1238 : :
1239 [ # # # # : 0 : ereport(ERROR,
# # ]
1240 : : (errcode_for_file_access(),
1241 : : errmsg("could not open file \"%s\" for writing: %m",
1242 : : cstate->filename),
1243 : : (save_errno == ENOENT || save_errno == EACCES) ?
1244 : : errhint("COPY TO instructs the PostgreSQL server process to write a file. "
1245 : : "You may want a client-side facility such as psql's \\copy.") : 0));
1246 : : }
1247 : :
1248 [ - + ]: 57 : if (fstat(fileno(cstate->copy_file), &st))
1249 [ # # ]: 0 : ereport(ERROR,
1250 : : (errcode_for_file_access(),
1251 : : errmsg("could not stat file \"%s\": %m",
1252 : : cstate->filename)));
1253 : :
1254 [ - + ]: 57 : if (S_ISDIR(st.st_mode))
1255 [ # # ]: 0 : ereport(ERROR,
1256 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1257 : : errmsg("\"%s\" is a directory", cstate->filename)));
1258 : : }
1259 : : }
1260 : :
1261 : : /* initialize progress */
1262 : 5099 : pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
1263 [ + + ]: 5099 : cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
1264 : 5099 : pgstat_progress_update_multi_param(2, progress_cols, progress_vals);
1265 : :
1266 : 5099 : cstate->bytes_processed = 0;
1267 : :
1268 : 5099 : MemoryContextSwitchTo(oldcontext);
1269 : :
1270 : 5099 : return cstate;
1271 : : }
1272 : :
1273 : : /*
1274 : : * Clean up storage and release resources for COPY TO.
1275 : : */
1276 : : void
1277 : 5097 : EndCopyTo(CopyToState cstate)
1278 : : {
1279 [ + + ]: 5097 : if (cstate->queryDesc != NULL)
1280 : : {
1281 : : /* Close down the query and free resources. */
1282 : 332 : ExecutorFinish(cstate->queryDesc);
1283 : 332 : ExecutorEnd(cstate->queryDesc);
1284 : 332 : FreeQueryDesc(cstate->queryDesc);
1285 : 332 : PopActiveSnapshot();
1286 : : }
1287 : :
1288 : : /* Clean up storage */
1289 : 5097 : EndCopy(cstate);
1290 : 5097 : }
1291 : :
1292 : : /*
1293 : : * Copy from relation or query TO file.
1294 : : *
1295 : : * Returns the number of rows processed.
1296 : : */
1297 : : uint64
1298 : 5099 : DoCopyTo(CopyToState cstate)
1299 : : {
1300 [ + + + + ]: 5099 : bool pipe = (cstate->filename == NULL && cstate->data_dest_cb == NULL);
1301 [ + + + - ]: 5099 : bool fe_copy = (pipe && whereToSendOutput == DestRemote);
1302 : : TupleDesc tupDesc;
1303 : : int num_phys_attrs;
1304 : : ListCell *cur;
1305 : 5099 : uint64 processed = 0;
1306 : :
1307 [ + + ]: 5099 : if (fe_copy)
1308 : 5041 : SendCopyBegin(cstate);
1309 : :
1310 [ + + ]: 5099 : if (cstate->rel)
1311 : 4767 : tupDesc = RelationGetDescr(cstate->rel);
1312 : : else
1313 : 332 : tupDesc = cstate->queryDesc->tupDesc;
1314 : 5099 : num_phys_attrs = tupDesc->natts;
1315 : 5099 : cstate->opts.null_print_client = cstate->opts.null_print; /* default */
1316 : :
1317 : : /* We use fe_msgbuf as a per-row buffer regardless of copy_dest */
1318 : 5099 : cstate->fe_msgbuf = makeStringInfo();
1319 : :
1320 : : /* Get info about the columns we need to process. */
1321 : 5099 : cstate->out_functions = palloc_array(FmgrInfo, num_phys_attrs);
1322 [ + + + + : 22117 : foreach(cur, cstate->attnumlist)
+ + ]
1323 : : {
1324 : 17019 : int attnum = lfirst_int(cur);
1325 : 17019 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1326 : :
1327 : 17019 : cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
1328 : 17019 : &cstate->out_functions[attnum - 1]);
1329 : : }
1330 : :
1331 : : /*
1332 : : * Create a temporary memory context that we can reset once per row to
1333 : : * recover palloc'd memory. This avoids any problems with leaks inside
1334 : : * datatype output routines, and should be faster than retail pfree's
1335 : : * anyway. (We don't need a whole econtext as CopyFrom does.)
1336 : : */
1337 : 5098 : cstate->rowcontext = AllocSetContextCreate(CurrentMemoryContext,
1338 : : "COPY TO",
1339 : : ALLOCSET_DEFAULT_SIZES);
1340 : :
1341 : 5098 : cstate->routine->CopyToStart(cstate, tupDesc);
1342 : :
1343 [ + + ]: 5098 : if (cstate->rel)
1344 : : {
1345 : : /*
1346 : : * If COPY TO source table is a partitioned table, then open each
1347 : : * partition and process each individual partition.
1348 : : */
1349 [ + + ]: 4766 : if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
1350 : : {
1351 [ + - + + : 87 : foreach_oid(child, cstate->partitions)
+ + ]
1352 : : {
1353 : : Relation scan_rel;
1354 : :
1355 : : /* We already got the lock in BeginCopyTo */
1356 : 39 : scan_rel = table_open(child, NoLock);
1357 : 39 : CopyRelationTo(cstate, scan_rel, cstate->rel, &processed);
1358 : 39 : table_close(scan_rel, NoLock);
1359 : : }
1360 : : }
1361 : : else
1362 : 4742 : CopyRelationTo(cstate, cstate->rel, NULL, &processed);
1363 : : }
1364 : : else
1365 : : {
1366 : : /* run the plan --- the dest receiver will send tuples */
1367 : 332 : ExecutorRun(cstate->queryDesc, ForwardScanDirection, 0);
1368 : 332 : processed = ((DR_copy *) cstate->queryDesc->dest)->processed;
1369 : : }
1370 : :
1371 : 5097 : cstate->routine->CopyToEnd(cstate);
1372 : :
1373 : 5097 : MemoryContextDelete(cstate->rowcontext);
1374 : :
1375 [ + + ]: 5097 : if (fe_copy)
1376 : 5039 : SendCopyEnd(cstate);
1377 : :
1378 : 5097 : return processed;
1379 : : }
1380 : :
1381 : : /*
1382 : : * Scans a single table and exports its rows to the COPY destination.
1383 : : *
1384 : : * root_rel can be set to the root table of rel if rel is a partition
1385 : : * table so that we can send tuples in root_rel's rowtype, which might
1386 : : * differ from individual partitions.
1387 : : */
1388 : : static void
1389 : 4781 : CopyRelationTo(CopyToState cstate, Relation rel, Relation root_rel, uint64 *processed)
1390 : : {
1391 : : TupleTableSlot *slot;
1392 : : TableScanDesc scandesc;
1393 : 4781 : AttrMap *map = NULL;
1394 : 4781 : TupleTableSlot *root_slot = NULL;
1395 : :
1396 : 4781 : scandesc = table_beginscan(rel, GetActiveSnapshot(), 0, NULL,
1397 : : SO_NONE);
1398 : 4780 : slot = table_slot_create(rel, NULL);
1399 : :
1400 : : /*
1401 : : * If we are exporting partition data here, we check if converting tuples
1402 : : * to the root table's rowtype, because a partition might have column
1403 : : * order different than its root table.
1404 : : */
1405 [ + + ]: 4780 : if (root_rel != NULL)
1406 : : {
1407 : 39 : root_slot = table_slot_create(root_rel, NULL);
1408 : 39 : map = build_attrmap_by_name_if_req(RelationGetDescr(rel),
1409 : : RelationGetDescr(root_rel),
1410 : : false);
1411 : : }
1412 : :
1413 [ + + ]: 1839487 : while (table_scan_getnextslot(scandesc, ForwardScanDirection, slot))
1414 : : {
1415 : : TupleTableSlot *copyslot;
1416 : :
1417 [ - + ]: 1834707 : CHECK_FOR_INTERRUPTS();
1418 : :
1419 [ + + ]: 1834707 : if (map != NULL)
1420 : 26 : copyslot = execute_attr_map_slot(map, slot, root_slot);
1421 : : else
1422 : : {
1423 : : /* Deconstruct the tuple */
1424 : 1834681 : slot_getallattrs(slot);
1425 : 1834681 : copyslot = slot;
1426 : : }
1427 : :
1428 : : /* Format and send the data */
1429 : 1834707 : CopyOneRowTo(cstate, copyslot);
1430 : :
1431 : : /*
1432 : : * Increment the number of processed tuples, and report the progress.
1433 : : */
1434 : 1834707 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
1435 : 1834707 : ++(*processed));
1436 : : }
1437 : :
1438 : 4780 : ExecDropSingleTupleTableSlot(slot);
1439 : :
1440 [ + + ]: 4780 : if (root_slot != NULL)
1441 : 39 : ExecDropSingleTupleTableSlot(root_slot);
1442 : :
1443 [ + + ]: 4780 : if (map != NULL)
1444 : 11 : free_attrmap(map);
1445 : :
1446 : 4780 : table_endscan(scandesc);
1447 : 4780 : }
1448 : :
1449 : : /*
1450 : : * Emit one row during DoCopyTo().
1451 : : */
1452 : : static inline void
1453 : 1839505 : CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
1454 : : {
1455 : : MemoryContext oldcontext;
1456 : :
1457 : 1839505 : MemoryContextReset(cstate->rowcontext);
1458 : 1839505 : oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
1459 : :
1460 : : /* Make sure the tuple is fully deconstructed */
1461 : 1839505 : slot_getallattrs(slot);
1462 : :
1463 : 1839505 : cstate->routine->CopyToOneRow(cstate, slot);
1464 : :
1465 : 1839505 : MemoryContextSwitchTo(oldcontext);
1466 : 1839505 : }
1467 : :
1468 : : /*
1469 : : * Send text representation of one attribute, with conversion and escaping
1470 : : */
1471 : : #define DUMPSOFAR() \
1472 : : do { \
1473 : : if (ptr > start) \
1474 : : CopySendData(cstate, start, ptr - start); \
1475 : : } while (0)
1476 : :
1477 : : static void
1478 : 6550534 : CopyAttributeOutText(CopyToState cstate, const char *string)
1479 : : {
1480 : : const char *ptr;
1481 : : const char *start;
1482 : : char c;
1483 : 6550534 : char delimc = cstate->opts.delim[0];
1484 : :
1485 [ + + ]: 6550534 : if (cstate->need_transcoding)
1486 : 4 : ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding);
1487 : : else
1488 : 6550530 : ptr = string;
1489 : :
1490 : : /*
1491 : : * We have to grovel through the string searching for control characters
1492 : : * and instances of the delimiter character. In most cases, though, these
1493 : : * are infrequent. To avoid overhead from calling CopySendData once per
1494 : : * character, we dump out all characters between escaped characters in a
1495 : : * single call. The loop invariant is that the data from "start" to "ptr"
1496 : : * can be sent literally, but hasn't yet been.
1497 : : *
1498 : : * We can skip pg_encoding_mblen() overhead when encoding is safe, because
1499 : : * in valid backend encodings, extra bytes of a multibyte character never
1500 : : * look like ASCII. This loop is sufficiently performance-critical that
1501 : : * it's worth making two copies of it to get the IS_HIGHBIT_SET() test out
1502 : : * of the normal safe-encoding path.
1503 : : */
1504 [ + + ]: 6550534 : if (cstate->encoding_embeds_ascii)
1505 : : {
1506 : 4 : start = ptr;
1507 [ + + ]: 12 : while ((c = *ptr) != '\0')
1508 : : {
1509 [ - + ]: 8 : if ((unsigned char) c < (unsigned char) 0x20)
1510 : : {
1511 : : /*
1512 : : * \r and \n must be escaped, the others are traditional. We
1513 : : * prefer to dump these using the C-like notation, rather than
1514 : : * a backslash and the literal character, because it makes the
1515 : : * dump file a bit more proof against Microsoftish data
1516 : : * mangling.
1517 : : */
1518 [ # # # # : 0 : switch (c)
# # # ]
1519 : : {
1520 : 0 : case '\b':
1521 : 0 : c = 'b';
1522 : 0 : break;
1523 : 0 : case '\f':
1524 : 0 : c = 'f';
1525 : 0 : break;
1526 : 0 : case '\n':
1527 : 0 : c = 'n';
1528 : 0 : break;
1529 : 0 : case '\r':
1530 : 0 : c = 'r';
1531 : 0 : break;
1532 : 0 : case '\t':
1533 : 0 : c = 't';
1534 : 0 : break;
1535 : 0 : case '\v':
1536 : 0 : c = 'v';
1537 : 0 : break;
1538 : 0 : default:
1539 : : /* If it's the delimiter, must backslash it */
1540 [ # # ]: 0 : if (c == delimc)
1541 : 0 : break;
1542 : : /* All ASCII control chars are length 1 */
1543 : 0 : ptr++;
1544 : 0 : continue; /* fall to end of loop */
1545 : : }
1546 : : /* if we get here, we need to convert the control char */
1547 [ # # ]: 0 : DUMPSOFAR();
1548 : 0 : CopySendChar(cstate, '\\');
1549 : 0 : CopySendChar(cstate, c);
1550 : 0 : start = ++ptr; /* do not include char in next run */
1551 : : }
1552 [ + - - + ]: 8 : else if (c == '\\' || c == delimc)
1553 : : {
1554 [ # # ]: 0 : DUMPSOFAR();
1555 : 0 : CopySendChar(cstate, '\\');
1556 : 0 : start = ptr++; /* we include char in next run */
1557 : : }
1558 [ + + ]: 8 : else if (IS_HIGHBIT_SET(c))
1559 : 4 : ptr += pg_encoding_mblen(cstate->file_encoding, ptr);
1560 : : else
1561 : 4 : ptr++;
1562 : : }
1563 : : }
1564 : : else
1565 : : {
1566 : 6550530 : start = ptr;
1567 [ + + ]: 77206209 : while ((c = *ptr) != '\0')
1568 : : {
1569 [ + + ]: 70655679 : if ((unsigned char) c < (unsigned char) 0x20)
1570 : : {
1571 : : /*
1572 : : * \r and \n must be escaped, the others are traditional. We
1573 : : * prefer to dump these using the C-like notation, rather than
1574 : : * a backslash and the literal character, because it makes the
1575 : : * dump file a bit more proof against Microsoftish data
1576 : : * mangling.
1577 : : */
1578 [ - - + - : 6991 : switch (c)
+ - - ]
1579 : : {
1580 : 0 : case '\b':
1581 : 0 : c = 'b';
1582 : 0 : break;
1583 : 0 : case '\f':
1584 : 0 : c = 'f';
1585 : 0 : break;
1586 : 5917 : case '\n':
1587 : 5917 : c = 'n';
1588 : 5917 : break;
1589 : 0 : case '\r':
1590 : 0 : c = 'r';
1591 : 0 : break;
1592 : 1074 : case '\t':
1593 : 1074 : c = 't';
1594 : 1074 : break;
1595 : 0 : case '\v':
1596 : 0 : c = 'v';
1597 : 0 : break;
1598 : 0 : default:
1599 : : /* If it's the delimiter, must backslash it */
1600 [ # # ]: 0 : if (c == delimc)
1601 : 0 : break;
1602 : : /* All ASCII control chars are length 1 */
1603 : 0 : ptr++;
1604 : 0 : continue; /* fall to end of loop */
1605 : : }
1606 : : /* if we get here, we need to convert the control char */
1607 [ + + ]: 6991 : DUMPSOFAR();
1608 : 6991 : CopySendChar(cstate, '\\');
1609 : 6991 : CopySendChar(cstate, c);
1610 : 6991 : start = ++ptr; /* do not include char in next run */
1611 : : }
1612 [ + + - + ]: 70648688 : else if (c == '\\' || c == delimc)
1613 : : {
1614 [ + + ]: 2321 : DUMPSOFAR();
1615 : 2321 : CopySendChar(cstate, '\\');
1616 : 2321 : start = ptr++; /* we include char in next run */
1617 : : }
1618 : : else
1619 : 70646367 : ptr++;
1620 : : }
1621 : : }
1622 : :
1623 [ + + ]: 6550534 : DUMPSOFAR();
1624 : 6550534 : }
1625 : :
1626 : : /*
1627 : : * Send text representation of one attribute, with conversion and
1628 : : * CSV-style escaping
1629 : : */
1630 : : static void
1631 : 428 : CopyAttributeOutCSV(CopyToState cstate, const char *string,
1632 : : bool use_quote)
1633 : : {
1634 : : const char *ptr;
1635 : : const char *start;
1636 : : char c;
1637 : 428 : char delimc = cstate->opts.delim[0];
1638 : 428 : char quotec = cstate->opts.quote[0];
1639 : 428 : char escapec = cstate->opts.escape[0];
1640 : 428 : bool single_attr = (list_length(cstate->attnumlist) == 1);
1641 : :
1642 : : /* force quoting if it matches null_print (before conversion!) */
1643 [ + + + + ]: 428 : if (!use_quote && strcmp(string, cstate->opts.null_print) == 0)
1644 : 36 : use_quote = true;
1645 : :
1646 [ + + ]: 428 : if (cstate->need_transcoding)
1647 : 4 : ptr = pg_server_to_any(string, strlen(string), cstate->file_encoding);
1648 : : else
1649 : 424 : ptr = string;
1650 : :
1651 : : /*
1652 : : * Make a preliminary pass to discover if it needs quoting
1653 : : */
1654 [ + + ]: 428 : if (!use_quote)
1655 : : {
1656 : : /*
1657 : : * Quote '\.' if it appears alone on a line, so that it will not be
1658 : : * interpreted as an end-of-data marker. (PG 18 and up will not
1659 : : * interpret '\.' in CSV that way, except in embedded-in-SQL data; but
1660 : : * we want the data to be loadable by older versions too. Also, this
1661 : : * avoids breaking clients that are still using PQgetline().)
1662 : : */
1663 [ + + + + ]: 304 : if (single_attr && strcmp(ptr, "\\.") == 0)
1664 : 4 : use_quote = true;
1665 : : else
1666 : : {
1667 : 300 : const char *tptr = ptr;
1668 : :
1669 [ + + ]: 1504 : while ((c = *tptr) != '\0')
1670 : : {
1671 [ + + + + : 1296 : if (c == delimc || c == quotec || c == '\n' || c == '\r')
+ + + + ]
1672 : : {
1673 : 92 : use_quote = true;
1674 : 92 : break;
1675 : : }
1676 [ + + + + ]: 1204 : if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii)
1677 : 4 : tptr += pg_encoding_mblen(cstate->file_encoding, tptr);
1678 : : else
1679 : 1200 : tptr++;
1680 : : }
1681 : : }
1682 : : }
1683 : :
1684 [ + + ]: 428 : if (use_quote)
1685 : : {
1686 : 220 : CopySendChar(cstate, quotec);
1687 : :
1688 : : /*
1689 : : * We adopt the same optimization strategy as in CopyAttributeOutText
1690 : : */
1691 : 220 : start = ptr;
1692 [ + + ]: 1704 : while ((c = *ptr) != '\0')
1693 : : {
1694 [ + + + + ]: 1484 : if (c == quotec || c == escapec)
1695 : : {
1696 [ + + ]: 104 : DUMPSOFAR();
1697 : 104 : CopySendChar(cstate, escapec);
1698 : 104 : start = ptr; /* we include char in next run */
1699 : : }
1700 [ + + + - ]: 1484 : if (IS_HIGHBIT_SET(c) && cstate->encoding_embeds_ascii)
1701 : 4 : ptr += pg_encoding_mblen(cstate->file_encoding, ptr);
1702 : : else
1703 : 1480 : ptr++;
1704 : : }
1705 [ + + ]: 220 : DUMPSOFAR();
1706 : :
1707 : 220 : CopySendChar(cstate, quotec);
1708 : : }
1709 : : else
1710 : : {
1711 : : /* If it doesn't need quoting, we can just dump it as-is */
1712 : 208 : CopySendString(cstate, ptr);
1713 : : }
1714 : 428 : }
1715 : :
1716 : : /*
1717 : : * copy_dest_startup --- executor startup
1718 : : */
1719 : : static void
1720 : 332 : copy_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
1721 : : {
1722 : : /* no-op */
1723 : 332 : }
1724 : :
1725 : : /*
1726 : : * copy_dest_receive --- receive one tuple
1727 : : */
1728 : : static bool
1729 : 4798 : copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
1730 : : {
1731 : 4798 : DR_copy *myState = (DR_copy *) self;
1732 : 4798 : CopyToState cstate = myState->cstate;
1733 : :
1734 : : /* Send the data */
1735 : 4798 : CopyOneRowTo(cstate, slot);
1736 : :
1737 : : /* Increment the number of processed tuples, and report the progress */
1738 : 4798 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
1739 : 4798 : ++myState->processed);
1740 : :
1741 : 4798 : return true;
1742 : : }
1743 : :
1744 : : /*
1745 : : * copy_dest_shutdown --- executor end
1746 : : */
1747 : : static void
1748 : 332 : copy_dest_shutdown(DestReceiver *self)
1749 : : {
1750 : : /* no-op */
1751 : 332 : }
1752 : :
1753 : : /*
1754 : : * copy_dest_destroy --- release DestReceiver object
1755 : : */
1756 : : static void
1757 : 0 : copy_dest_destroy(DestReceiver *self)
1758 : : {
1759 : 0 : pfree(self);
1760 : 0 : }
1761 : :
1762 : : /*
1763 : : * CreateCopyDestReceiver -- create a suitable DestReceiver object
1764 : : */
1765 : : DestReceiver *
1766 : 336 : CreateCopyDestReceiver(void)
1767 : : {
1768 : 336 : DR_copy *self = palloc_object(DR_copy);
1769 : :
1770 : 336 : self->pub.receiveSlot = copy_dest_receive;
1771 : 336 : self->pub.rStartup = copy_dest_startup;
1772 : 336 : self->pub.rShutdown = copy_dest_shutdown;
1773 : 336 : self->pub.rDestroy = copy_dest_destroy;
1774 : 336 : self->pub.mydest = DestCopyOut;
1775 : :
1776 : 336 : self->cstate = NULL; /* will be set later */
1777 : 336 : self->processed = 0;
1778 : :
1779 : 336 : return (DestReceiver *) self;
1780 : : }
|
