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