Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * copyfrom.c
4 : : * COPY <table> FROM file/program/client
5 : : *
6 : : * This file contains routines needed to efficiently load tuples into a
7 : : * table. That includes looking up the correct partition, firing triggers,
8 : : * calling the table AM function to insert the data, and updating indexes.
9 : : * Reading data from the input file or client and parsing it into Datums
10 : : * is handled in copyfromparse.c.
11 : : *
12 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
13 : : * Portions Copyright (c) 1994, Regents of the University of California
14 : : *
15 : : *
16 : : * IDENTIFICATION
17 : : * src/backend/commands/copyfrom.c
18 : : *
19 : : *-------------------------------------------------------------------------
20 : : */
21 : : #include "postgres.h"
22 : :
23 : : #include <ctype.h>
24 : : #include <unistd.h>
25 : : #include <sys/stat.h>
26 : :
27 : : #include "access/heapam.h"
28 : : #include "access/tableam.h"
29 : : #include "access/tupconvert.h"
30 : : #include "access/xact.h"
31 : : #include "catalog/namespace.h"
32 : : #include "commands/copyapi.h"
33 : : #include "commands/copyfrom_internal.h"
34 : : #include "commands/progress.h"
35 : : #include "commands/trigger.h"
36 : : #include "executor/execPartition.h"
37 : : #include "executor/executor.h"
38 : : #include "executor/nodeModifyTable.h"
39 : : #include "executor/tuptable.h"
40 : : #include "foreign/fdwapi.h"
41 : : #include "mb/pg_wchar.h"
42 : : #include "miscadmin.h"
43 : : #include "nodes/miscnodes.h"
44 : : #include "optimizer/optimizer.h"
45 : : #include "pgstat.h"
46 : : #include "rewrite/rewriteHandler.h"
47 : : #include "storage/fd.h"
48 : : #include "tcop/tcopprot.h"
49 : : #include "utils/lsyscache.h"
50 : : #include "utils/memutils.h"
51 : : #include "utils/portal.h"
52 : : #include "utils/rel.h"
53 : : #include "utils/snapmgr.h"
54 : : #include "utils/typcache.h"
55 : :
56 : : /*
57 : : * No more than this many tuples per CopyMultiInsertBuffer
58 : : *
59 : : * Caution: Don't make this too big, as we could end up with this many
60 : : * CopyMultiInsertBuffer items stored in CopyMultiInsertInfo's
61 : : * multiInsertBuffers list. Increasing this can cause quadratic growth in
62 : : * memory requirements during copies into partitioned tables with a large
63 : : * number of partitions.
64 : : */
65 : : #define MAX_BUFFERED_TUPLES 1000
66 : :
67 : : /*
68 : : * Flush buffers if there are >= this many bytes, as counted by the input
69 : : * size, of tuples stored.
70 : : */
71 : : #define MAX_BUFFERED_BYTES 65535
72 : :
73 : : /*
74 : : * Trim the list of buffers back down to this number after flushing. This
75 : : * must be >= 2.
76 : : */
77 : : #define MAX_PARTITION_BUFFERS 32
78 : :
79 : : /* Stores multi-insert data related to a single relation in CopyFrom. */
80 : : typedef struct CopyMultiInsertBuffer
81 : : {
82 : : TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */
83 : : ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */
84 : : BulkInsertState bistate; /* BulkInsertState for this rel if plain
85 : : * table; NULL if foreign table */
86 : : int nused; /* number of 'slots' containing tuples */
87 : : uint64 linenos[MAX_BUFFERED_TUPLES]; /* Line # of tuple in copy
88 : : * stream */
89 : : } CopyMultiInsertBuffer;
90 : :
91 : : /*
92 : : * Stores one or many CopyMultiInsertBuffers and details about the size and
93 : : * number of tuples which are stored in them. This allows multiple buffers to
94 : : * exist at once when COPYing into a partitioned table.
95 : : */
96 : : typedef struct CopyMultiInsertInfo
97 : : {
98 : : List *multiInsertBuffers; /* List of tracked CopyMultiInsertBuffers */
99 : : int bufferedTuples; /* number of tuples buffered over all buffers */
100 : : int bufferedBytes; /* number of bytes from all buffered tuples */
101 : : CopyFromState cstate; /* Copy state for this CopyMultiInsertInfo */
102 : : EState *estate; /* Executor state used for COPY */
103 : : CommandId mycid; /* Command Id used for COPY */
104 : : uint32 ti_options; /* table insert options */
105 : : } CopyMultiInsertInfo;
106 : :
107 : :
108 : : /* non-export function prototypes */
109 : : static void ClosePipeFromProgram(CopyFromState cstate);
110 : :
111 : : /*
112 : : * Built-in format-specific routines. One-row callbacks are defined in
113 : : * copyfromparse.c.
114 : : */
115 : : static void CopyFromTextLikeInFunc(CopyFromState cstate, Oid atttypid, FmgrInfo *finfo,
116 : : Oid *typioparam);
117 : : static void CopyFromTextLikeStart(CopyFromState cstate, TupleDesc tupDesc);
118 : : static void CopyFromTextLikeEnd(CopyFromState cstate);
119 : : static void CopyFromBinaryInFunc(CopyFromState cstate, Oid atttypid,
120 : : FmgrInfo *finfo, Oid *typioparam);
121 : : static void CopyFromBinaryStart(CopyFromState cstate, TupleDesc tupDesc);
122 : : static void CopyFromBinaryEnd(CopyFromState cstate);
123 : :
124 : :
125 : : /*
126 : : * COPY FROM routines for built-in formats.
127 : : *
128 : : * CSV and text formats share the same TextLike routines except for the
129 : : * one-row callback.
130 : : */
131 : :
132 : : /* text format */
133 : : static const CopyFromRoutine CopyFromRoutineText = {
134 : : .CopyFromInFunc = CopyFromTextLikeInFunc,
135 : : .CopyFromStart = CopyFromTextLikeStart,
136 : : .CopyFromOneRow = CopyFromTextOneRow,
137 : : .CopyFromEnd = CopyFromTextLikeEnd,
138 : : };
139 : :
140 : : /* CSV format */
141 : : static const CopyFromRoutine CopyFromRoutineCSV = {
142 : : .CopyFromInFunc = CopyFromTextLikeInFunc,
143 : : .CopyFromStart = CopyFromTextLikeStart,
144 : : .CopyFromOneRow = CopyFromCSVOneRow,
145 : : .CopyFromEnd = CopyFromTextLikeEnd,
146 : : };
147 : :
148 : : /* binary format */
149 : : static const CopyFromRoutine CopyFromRoutineBinary = {
150 : : .CopyFromInFunc = CopyFromBinaryInFunc,
151 : : .CopyFromStart = CopyFromBinaryStart,
152 : : .CopyFromOneRow = CopyFromBinaryOneRow,
153 : : .CopyFromEnd = CopyFromBinaryEnd,
154 : : };
155 : :
156 : : /* Return a COPY FROM routine for the given options */
157 : : static const CopyFromRoutine *
158 : 1201 : CopyFromGetRoutine(const CopyFormatOptions *opts)
159 : : {
160 [ + + ]: 1201 : if (opts->format == COPY_FORMAT_CSV)
161 : 189 : return &CopyFromRoutineCSV;
162 [ + + ]: 1012 : else if (opts->format == COPY_FORMAT_BINARY)
163 : 9 : return &CopyFromRoutineBinary;
164 : :
165 : : /* default is text */
166 : 1003 : return &CopyFromRoutineText;
167 : : }
168 : :
169 : : /* Implementation of the start callback for text and CSV formats */
170 : : static void
171 : 1176 : CopyFromTextLikeStart(CopyFromState cstate, TupleDesc tupDesc)
172 : : {
173 : : AttrNumber attr_count;
174 : :
175 : : /*
176 : : * If encoding conversion is needed, we need another buffer to hold the
177 : : * converted input data. Otherwise, we can just point input_buf to the
178 : : * same buffer as raw_buf.
179 : : */
180 [ + + ]: 1176 : if (cstate->need_transcoding)
181 : : {
182 : 24 : cstate->input_buf = (char *) palloc(INPUT_BUF_SIZE + 1);
183 : 24 : cstate->input_buf_index = cstate->input_buf_len = 0;
184 : : }
185 : : else
186 : 1152 : cstate->input_buf = cstate->raw_buf;
187 : 1176 : cstate->input_reached_eof = false;
188 : :
189 : 1176 : initStringInfo(&cstate->line_buf);
190 : :
191 : : /*
192 : : * Create workspace for CopyReadAttributes results; used by CSV and text
193 : : * format.
194 : : */
195 : 1176 : attr_count = list_length(cstate->attnumlist);
196 : 1176 : cstate->max_fields = attr_count;
197 : 1176 : cstate->raw_fields = (char **) palloc(attr_count * sizeof(char *));
198 : 1176 : }
199 : :
200 : : /*
201 : : * Implementation of the infunc callback for text and CSV formats. Assign
202 : : * the input function data to the given *finfo.
203 : : */
204 : : static void
205 : 3454 : CopyFromTextLikeInFunc(CopyFromState cstate, Oid atttypid, FmgrInfo *finfo,
206 : : Oid *typioparam)
207 : : {
208 : : Oid func_oid;
209 : :
210 : 3454 : getTypeInputInfo(atttypid, &func_oid, typioparam);
211 : 3454 : fmgr_info(func_oid, finfo);
212 : 3454 : }
213 : :
214 : : /* Implementation of the end callback for text and CSV formats */
215 : : static void
216 : 984 : CopyFromTextLikeEnd(CopyFromState cstate)
217 : : {
218 : : /* nothing to do */
219 : 984 : }
220 : :
221 : : /* Implementation of the start callback for binary format */
222 : : static void
223 : 8 : CopyFromBinaryStart(CopyFromState cstate, TupleDesc tupDesc)
224 : : {
225 : : /* Read and verify binary header */
226 : 8 : ReceiveCopyBinaryHeader(cstate);
227 : 8 : }
228 : :
229 : : /*
230 : : * Implementation of the infunc callback for binary format. Assign
231 : : * the binary input function to the given *finfo.
232 : : */
233 : : static void
234 : 38 : CopyFromBinaryInFunc(CopyFromState cstate, Oid atttypid,
235 : : FmgrInfo *finfo, Oid *typioparam)
236 : : {
237 : : Oid func_oid;
238 : :
239 : 38 : getTypeBinaryInputInfo(atttypid, &func_oid, typioparam);
240 : 37 : fmgr_info(func_oid, finfo);
241 : 37 : }
242 : :
243 : : /* Implementation of the end callback for binary format */
244 : : static void
245 : 7 : CopyFromBinaryEnd(CopyFromState cstate)
246 : : {
247 : : /* nothing to do */
248 : 7 : }
249 : :
250 : : /*
251 : : * error context callback for COPY FROM
252 : : *
253 : : * The argument for the error context must be CopyFromState.
254 : : */
255 : : void
256 : 248 : CopyFromErrorCallback(void *arg)
257 : : {
258 : 248 : CopyFromState cstate = (CopyFromState) arg;
259 : :
260 [ + + ]: 248 : if (cstate->relname_only)
261 : : {
262 : 45 : errcontext("COPY %s",
263 : : cstate->cur_relname);
264 : 45 : return;
265 : : }
266 [ + + ]: 203 : if (cstate->opts.format == COPY_FORMAT_BINARY)
267 : : {
268 : : /* can't usefully display the data */
269 [ + - ]: 1 : if (cstate->cur_attname)
270 : 1 : errcontext("COPY %s, line %" PRIu64 ", column %s",
271 : : cstate->cur_relname,
272 : : cstate->cur_lineno,
273 : : cstate->cur_attname);
274 : : else
275 : 0 : errcontext("COPY %s, line %" PRIu64,
276 : : cstate->cur_relname,
277 : : cstate->cur_lineno);
278 : : }
279 : : else
280 : : {
281 [ + + + + ]: 202 : if (cstate->cur_attname && cstate->cur_attval)
282 : 34 : {
283 : : /* error is relevant to a particular column */
284 : : char *attval;
285 : :
286 : 34 : attval = CopyLimitPrintoutLength(cstate->cur_attval);
287 : 34 : errcontext("COPY %s, line %" PRIu64 ", column %s: \"%s\"",
288 : : cstate->cur_relname,
289 : : cstate->cur_lineno,
290 : : cstate->cur_attname,
291 : : attval);
292 : 34 : pfree(attval);
293 : : }
294 [ + + ]: 168 : else if (cstate->cur_attname)
295 : : {
296 : : /* error is relevant to a particular column, value is NULL */
297 : 12 : errcontext("COPY %s, line %" PRIu64 ", column %s: null input",
298 : : cstate->cur_relname,
299 : : cstate->cur_lineno,
300 : : cstate->cur_attname);
301 : : }
302 : : else
303 : : {
304 : : /*
305 : : * Error is relevant to a particular line.
306 : : *
307 : : * If line_buf still contains the correct line, print it.
308 : : */
309 [ + + ]: 156 : if (cstate->line_buf_valid)
310 : : {
311 : : char *lineval;
312 : :
313 : 129 : lineval = CopyLimitPrintoutLength(cstate->line_buf.data);
314 : 129 : errcontext("COPY %s, line %" PRIu64 ": \"%s\"",
315 : : cstate->cur_relname,
316 : : cstate->cur_lineno, lineval);
317 : 129 : pfree(lineval);
318 : : }
319 : : else
320 : : {
321 : 27 : errcontext("COPY %s, line %" PRIu64,
322 : : cstate->cur_relname,
323 : : cstate->cur_lineno);
324 : : }
325 : : }
326 : : }
327 : : }
328 : :
329 : : /*
330 : : * Make sure we don't print an unreasonable amount of COPY data in a message.
331 : : *
332 : : * Returns a pstrdup'd copy of the input.
333 : : */
334 : : char *
335 : 203 : CopyLimitPrintoutLength(const char *str)
336 : : {
337 : : #define MAX_COPY_DATA_DISPLAY 100
338 : :
339 : 203 : int slen = strlen(str);
340 : : int len;
341 : : char *res;
342 : :
343 : : /* Fast path if definitely okay */
344 [ + - ]: 203 : if (slen <= MAX_COPY_DATA_DISPLAY)
345 : 203 : return pstrdup(str);
346 : :
347 : : /* Apply encoding-dependent truncation */
348 : 0 : len = pg_mbcliplen(str, slen, MAX_COPY_DATA_DISPLAY);
349 : :
350 : : /*
351 : : * Truncate, and add "..." to show we truncated the input.
352 : : */
353 : 0 : res = (char *) palloc(len + 4);
354 : 0 : memcpy(res, str, len);
355 : 0 : strcpy(res + len, "...");
356 : :
357 : 0 : return res;
358 : : }
359 : :
360 : : /*
361 : : * Allocate memory and initialize a new CopyMultiInsertBuffer for this
362 : : * ResultRelInfo.
363 : : */
364 : : static CopyMultiInsertBuffer *
365 : 1030 : CopyMultiInsertBufferInit(ResultRelInfo *rri)
366 : : {
367 : : CopyMultiInsertBuffer *buffer;
368 : :
369 : 1030 : buffer = palloc_object(CopyMultiInsertBuffer);
370 : 1030 : memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES);
371 : 1030 : buffer->resultRelInfo = rri;
372 [ + + ]: 1030 : buffer->bistate = (rri->ri_FdwRoutine == NULL) ? GetBulkInsertState() : NULL;
373 : 1030 : buffer->nused = 0;
374 : :
375 : 1030 : return buffer;
376 : : }
377 : :
378 : : /*
379 : : * Make a new buffer for this ResultRelInfo.
380 : : */
381 : : static inline void
382 : 1030 : CopyMultiInsertInfoSetupBuffer(CopyMultiInsertInfo *miinfo,
383 : : ResultRelInfo *rri)
384 : : {
385 : : CopyMultiInsertBuffer *buffer;
386 : :
387 : 1030 : buffer = CopyMultiInsertBufferInit(rri);
388 : :
389 : : /* Setup back-link so we can easily find this buffer again */
390 : 1030 : rri->ri_CopyMultiInsertBuffer = buffer;
391 : : /* Record that we're tracking this buffer */
392 : 1030 : miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
393 : 1030 : }
394 : :
395 : : /*
396 : : * Initialize an already allocated CopyMultiInsertInfo.
397 : : *
398 : : * If rri is a non-partitioned table then a CopyMultiInsertBuffer is set up
399 : : * for that table.
400 : : */
401 : : static void
402 : 1026 : CopyMultiInsertInfoInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
403 : : CopyFromState cstate, EState *estate, CommandId mycid,
404 : : uint32 ti_options)
405 : : {
406 : 1026 : miinfo->multiInsertBuffers = NIL;
407 : 1026 : miinfo->bufferedTuples = 0;
408 : 1026 : miinfo->bufferedBytes = 0;
409 : 1026 : miinfo->cstate = cstate;
410 : 1026 : miinfo->estate = estate;
411 : 1026 : miinfo->mycid = mycid;
412 : 1026 : miinfo->ti_options = ti_options;
413 : :
414 : : /*
415 : : * Only setup the buffer when not dealing with a partitioned table.
416 : : * Buffers for partitioned tables will just be setup when we need to send
417 : : * tuples their way for the first time.
418 : : */
419 [ + + ]: 1026 : if (rri->ri_RelationDesc->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
420 : 981 : CopyMultiInsertInfoSetupBuffer(miinfo, rri);
421 : 1026 : }
422 : :
423 : : /*
424 : : * Returns true if the buffers are full
425 : : */
426 : : static inline bool
427 : 718172 : CopyMultiInsertInfoIsFull(CopyMultiInsertInfo *miinfo)
428 : : {
429 [ + + ]: 718172 : if (miinfo->bufferedTuples >= MAX_BUFFERED_TUPLES ||
430 [ + + ]: 717586 : miinfo->bufferedBytes >= MAX_BUFFERED_BYTES)
431 : 656 : return true;
432 : 717516 : return false;
433 : : }
434 : :
435 : : /*
436 : : * Returns true if we have no buffered tuples
437 : : */
438 : : static inline bool
439 : 908 : CopyMultiInsertInfoIsEmpty(CopyMultiInsertInfo *miinfo)
440 : : {
441 : 908 : return miinfo->bufferedTuples == 0;
442 : : }
443 : :
444 : : /*
445 : : * Write the tuples stored in 'buffer' out to the table.
446 : : */
447 : : static inline void
448 : 1482 : CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo,
449 : : CopyMultiInsertBuffer *buffer,
450 : : int64 *processed)
451 : : {
452 : 1482 : CopyFromState cstate = miinfo->cstate;
453 : 1482 : EState *estate = miinfo->estate;
454 : 1482 : int nused = buffer->nused;
455 : 1482 : ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
456 : 1482 : TupleTableSlot **slots = buffer->slots;
457 : : int i;
458 : :
459 [ + + ]: 1482 : if (resultRelInfo->ri_FdwRoutine)
460 : : {
461 : 7 : int batch_size = resultRelInfo->ri_BatchSize;
462 : 7 : int sent = 0;
463 : :
464 : : Assert(buffer->bistate == NULL);
465 : :
466 : : /* Ensure that the FDW supports batching and it's enabled */
467 : : Assert(resultRelInfo->ri_FdwRoutine->ExecForeignBatchInsert);
468 : : Assert(batch_size > 1);
469 : :
470 : : /*
471 : : * We suppress error context information other than the relation name,
472 : : * if one of the operations below fails.
473 : : */
474 : : Assert(!cstate->relname_only);
475 : 7 : cstate->relname_only = true;
476 : :
477 [ + + ]: 19 : while (sent < nused)
478 : : {
479 : 13 : int size = (batch_size < nused - sent) ? batch_size : (nused - sent);
480 : 13 : int inserted = size;
481 : : TupleTableSlot **rslots;
482 : :
483 : : /* insert into foreign table: let the FDW do it */
484 : : rslots =
485 : 13 : resultRelInfo->ri_FdwRoutine->ExecForeignBatchInsert(estate,
486 : : resultRelInfo,
487 : 13 : &slots[sent],
488 : : NULL,
489 : : &inserted);
490 : :
491 : 12 : sent += size;
492 : :
493 : : /* No need to do anything if there are no inserted rows */
494 [ + + ]: 12 : if (inserted <= 0)
495 : 2 : continue;
496 : :
497 : : /* Triggers on foreign tables should not have transition tables */
498 : : Assert(resultRelInfo->ri_TrigDesc == NULL ||
499 : : resultRelInfo->ri_TrigDesc->trig_insert_new_table == false);
500 : :
501 : : /* Run AFTER ROW INSERT triggers */
502 [ - + ]: 10 : if (resultRelInfo->ri_TrigDesc != NULL &&
503 [ # # ]: 0 : resultRelInfo->ri_TrigDesc->trig_insert_after_row)
504 : : {
505 : 0 : Oid relid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
506 : :
507 [ # # ]: 0 : for (i = 0; i < inserted; i++)
508 : : {
509 : 0 : TupleTableSlot *slot = rslots[i];
510 : :
511 : : /*
512 : : * AFTER ROW Triggers might reference the tableoid column,
513 : : * so (re-)initialize tts_tableOid before evaluating them.
514 : : */
515 : 0 : slot->tts_tableOid = relid;
516 : :
517 : 0 : ExecARInsertTriggers(estate, resultRelInfo,
518 : : slot, NIL,
519 : : cstate->transition_capture);
520 : : }
521 : : }
522 : :
523 : : /* Update the row counter and progress of the COPY command */
524 : 10 : *processed += inserted;
525 : 10 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
526 : : *processed);
527 : : }
528 : :
529 [ + + ]: 24 : for (i = 0; i < nused; i++)
530 : 18 : ExecClearTuple(slots[i]);
531 : :
532 : : /* reset relname_only */
533 : 6 : cstate->relname_only = false;
534 : : }
535 : : else
536 : : {
537 : 1475 : CommandId mycid = miinfo->mycid;
538 : 1475 : uint32 ti_options = miinfo->ti_options;
539 : 1475 : bool line_buf_valid = cstate->line_buf_valid;
540 : 1475 : uint64 save_cur_lineno = cstate->cur_lineno;
541 : : MemoryContext oldcontext;
542 : :
543 : : Assert(buffer->bistate != NULL);
544 : :
545 : : /*
546 : : * Print error context information correctly, if one of the operations
547 : : * below fails.
548 : : */
549 : 1475 : cstate->line_buf_valid = false;
550 : :
551 : : /*
552 : : * table_multi_insert may leak memory, so switch to short-lived memory
553 : : * context before calling it.
554 : : */
555 [ + - ]: 1475 : oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
556 : 1475 : table_multi_insert(resultRelInfo->ri_RelationDesc,
557 : : slots,
558 : : nused,
559 : : mycid,
560 : : ti_options,
561 : 1475 : buffer->bistate);
562 : 1475 : MemoryContextSwitchTo(oldcontext);
563 : :
564 [ + + ]: 719544 : for (i = 0; i < nused; i++)
565 : : {
566 : : /*
567 : : * If there are any indexes, update them for all the inserted
568 : : * tuples, and run AFTER ROW INSERT triggers.
569 : : */
570 [ + + ]: 718078 : if (resultRelInfo->ri_NumIndices > 0)
571 : : {
572 : : List *recheckIndexes;
573 : :
574 : 100992 : cstate->cur_lineno = buffer->linenos[i];
575 : : recheckIndexes =
576 : 100992 : ExecInsertIndexTuples(resultRelInfo,
577 : : estate, 0, buffer->slots[i],
578 : : NIL, NULL);
579 : 100983 : ExecARInsertTriggers(estate, resultRelInfo,
580 : 100983 : slots[i], recheckIndexes,
581 : : cstate->transition_capture);
582 : 100983 : list_free(recheckIndexes);
583 : : }
584 : :
585 : : /*
586 : : * There's no indexes, but see if we need to run AFTER ROW INSERT
587 : : * triggers anyway.
588 : : */
589 [ + + ]: 617086 : else if (resultRelInfo->ri_TrigDesc != NULL &&
590 [ + + ]: 56 : (resultRelInfo->ri_TrigDesc->trig_insert_after_row ||
591 [ + + ]: 44 : resultRelInfo->ri_TrigDesc->trig_insert_new_table))
592 : : {
593 : 24 : cstate->cur_lineno = buffer->linenos[i];
594 : 24 : ExecARInsertTriggers(estate, resultRelInfo,
595 : 24 : slots[i], NIL,
596 : : cstate->transition_capture);
597 : : }
598 : :
599 : 718069 : ExecClearTuple(slots[i]);
600 : : }
601 : :
602 : : /* Update the row counter and progress of the COPY command */
603 : 1466 : *processed += nused;
604 : 1466 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
605 : : *processed);
606 : :
607 : : /* reset cur_lineno and line_buf_valid to what they were */
608 : 1466 : cstate->line_buf_valid = line_buf_valid;
609 : 1466 : cstate->cur_lineno = save_cur_lineno;
610 : : }
611 : :
612 : : /* Mark that all slots are free */
613 : 1472 : buffer->nused = 0;
614 : 1472 : }
615 : :
616 : : /*
617 : : * Drop used slots and free member for this buffer.
618 : : *
619 : : * The buffer must be flushed before cleanup.
620 : : */
621 : : static inline void
622 : 884 : CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
623 : : CopyMultiInsertBuffer *buffer)
624 : : {
625 : 884 : ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
626 : : int i;
627 : :
628 : : /* Ensure buffer was flushed */
629 : : Assert(buffer->nused == 0);
630 : :
631 : : /* Remove back-link to ourself */
632 : 884 : resultRelInfo->ri_CopyMultiInsertBuffer = NULL;
633 : :
634 [ + + ]: 884 : if (resultRelInfo->ri_FdwRoutine == NULL)
635 : : {
636 : : Assert(buffer->bistate != NULL);
637 : 878 : FreeBulkInsertState(buffer->bistate);
638 : : }
639 : : else
640 : : Assert(buffer->bistate == NULL);
641 : :
642 : : /* Since we only create slots on demand, just drop the non-null ones. */
643 [ + + + + ]: 150833 : for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++)
644 : 149949 : ExecDropSingleTupleTableSlot(buffer->slots[i]);
645 : :
646 [ + + ]: 884 : if (resultRelInfo->ri_FdwRoutine == NULL)
647 : 878 : table_finish_bulk_insert(resultRelInfo->ri_RelationDesc,
648 : : miinfo->ti_options);
649 : :
650 : 884 : pfree(buffer);
651 : 884 : }
652 : :
653 : : /*
654 : : * Write out all stored tuples in all buffers out to the tables.
655 : : *
656 : : * Once flushed we also trim the tracked buffers list down to size by removing
657 : : * the buffers created earliest first.
658 : : *
659 : : * Callers should pass 'curr_rri' as the ResultRelInfo that's currently being
660 : : * used. When cleaning up old buffers we'll never remove the one for
661 : : * 'curr_rri'.
662 : : */
663 : : static inline void
664 : 1367 : CopyMultiInsertInfoFlush(CopyMultiInsertInfo *miinfo, ResultRelInfo *curr_rri,
665 : : int64 *processed)
666 : : {
667 : : ListCell *lc;
668 : :
669 [ + - + + : 2839 : foreach(lc, miinfo->multiInsertBuffers)
+ + ]
670 : : {
671 : 1482 : CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) lfirst(lc);
672 : :
673 : 1482 : CopyMultiInsertBufferFlush(miinfo, buffer, processed);
674 : : }
675 : :
676 : 1357 : miinfo->bufferedTuples = 0;
677 : 1357 : miinfo->bufferedBytes = 0;
678 : :
679 : : /*
680 : : * Trim the list of tracked buffers down if it exceeds the limit. Here we
681 : : * remove buffers starting with the ones we created first. It seems less
682 : : * likely that these older ones will be needed than the ones that were
683 : : * just created.
684 : : */
685 [ - + ]: 1357 : while (list_length(miinfo->multiInsertBuffers) > MAX_PARTITION_BUFFERS)
686 : : {
687 : : CopyMultiInsertBuffer *buffer;
688 : :
689 : 0 : buffer = (CopyMultiInsertBuffer *) linitial(miinfo->multiInsertBuffers);
690 : :
691 : : /*
692 : : * We never want to remove the buffer that's currently being used, so
693 : : * if we happen to find that then move it to the end of the list.
694 : : */
695 [ # # ]: 0 : if (buffer->resultRelInfo == curr_rri)
696 : : {
697 : : /*
698 : : * The code below would misbehave if we were trying to reduce the
699 : : * list to less than two items.
700 : : */
701 : : StaticAssertDecl(MAX_PARTITION_BUFFERS >= 2,
702 : : "MAX_PARTITION_BUFFERS must be >= 2");
703 : :
704 : 0 : miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
705 : 0 : miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
706 : 0 : buffer = (CopyMultiInsertBuffer *) linitial(miinfo->multiInsertBuffers);
707 : : }
708 : :
709 : 0 : CopyMultiInsertBufferCleanup(miinfo, buffer);
710 : 0 : miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
711 : : }
712 : 1357 : }
713 : :
714 : : /*
715 : : * Cleanup allocated buffers and free memory
716 : : */
717 : : static inline void
718 : 880 : CopyMultiInsertInfoCleanup(CopyMultiInsertInfo *miinfo)
719 : : {
720 : : ListCell *lc;
721 : :
722 [ + + + + : 1764 : foreach(lc, miinfo->multiInsertBuffers)
+ + ]
723 : 884 : CopyMultiInsertBufferCleanup(miinfo, lfirst(lc));
724 : :
725 : 880 : list_free(miinfo->multiInsertBuffers);
726 : 880 : }
727 : :
728 : : /*
729 : : * Get the next TupleTableSlot that the next tuple should be stored in.
730 : : *
731 : : * Callers must ensure that the buffer is not full.
732 : : *
733 : : * Note: 'miinfo' is unused but has been included for consistency with the
734 : : * other functions in this area.
735 : : */
736 : : static inline TupleTableSlot *
737 : 719225 : CopyMultiInsertInfoNextFreeSlot(CopyMultiInsertInfo *miinfo,
738 : : ResultRelInfo *rri)
739 : : {
740 : 719225 : CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer;
741 : : int nused;
742 : :
743 : : Assert(buffer != NULL);
744 : : Assert(buffer->nused < MAX_BUFFERED_TUPLES);
745 : :
746 : 719225 : nused = buffer->nused;
747 : :
748 [ + + ]: 719225 : if (buffer->slots[nused] == NULL)
749 : 150179 : buffer->slots[nused] = table_slot_create(rri->ri_RelationDesc, NULL);
750 : 719225 : return buffer->slots[nused];
751 : : }
752 : :
753 : : /*
754 : : * Record the previously reserved TupleTableSlot that was reserved by
755 : : * CopyMultiInsertInfoNextFreeSlot as being consumed.
756 : : */
757 : : static inline void
758 : 718172 : CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
759 : : TupleTableSlot *slot, int tuplen, uint64 lineno)
760 : : {
761 : 718172 : CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer;
762 : :
763 : : Assert(buffer != NULL);
764 : : Assert(slot == buffer->slots[buffer->nused]);
765 : :
766 : : /* Store the line number so we can properly report any errors later */
767 : 718172 : buffer->linenos[buffer->nused] = lineno;
768 : :
769 : : /* Record this slot as being used */
770 : 718172 : buffer->nused++;
771 : :
772 : : /* Update how many tuples are stored and their size */
773 : 718172 : miinfo->bufferedTuples++;
774 : 718172 : miinfo->bufferedBytes += tuplen;
775 : 718172 : }
776 : :
777 : : /*
778 : : * Copy FROM file to relation.
779 : : */
780 : : uint64
781 : 1146 : CopyFrom(CopyFromState cstate)
782 : : {
783 : : ResultRelInfo *resultRelInfo;
784 : : ResultRelInfo *target_resultRelInfo;
785 : 1146 : ResultRelInfo *prevResultRelInfo = NULL;
786 : 1146 : EState *estate = CreateExecutorState(); /* for ExecConstraints() */
787 : : ModifyTableState *mtstate;
788 : : ExprContext *econtext;
789 : 1146 : TupleTableSlot *singleslot = NULL;
790 : 1146 : MemoryContext oldcontext = CurrentMemoryContext;
791 : :
792 : 1146 : PartitionTupleRouting *proute = NULL;
793 : : ErrorContextCallback errcallback;
794 : 1146 : CommandId mycid = GetCurrentCommandId(true);
795 : 1146 : uint32 ti_options = 0; /* start with default options for insert */
796 : 1146 : BulkInsertState bistate = NULL;
797 : : CopyInsertMethod insertMethod;
798 : 1146 : CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
799 : 1146 : int64 processed = 0;
800 : 1146 : int64 excluded = 0;
801 : : bool has_before_insert_row_trig;
802 : : bool has_instead_insert_row_trig;
803 : 1146 : bool leafpart_use_multi_insert = false;
804 : :
805 : : Assert(cstate->rel);
806 : : Assert(list_length(cstate->range_table) == 1);
807 : :
808 : 1146 : if (cstate->opts.on_error != COPY_ON_ERROR_STOP)
809 : : Assert(cstate->escontext);
810 : :
811 : : /*
812 : : * The target must be a plain, foreign, or partitioned relation, or have
813 : : * an INSTEAD OF INSERT row trigger. (Currently, such triggers are only
814 : : * allowed on views, so we only hint about them in the view case.)
815 : : */
816 [ + + ]: 1146 : if (cstate->rel->rd_rel->relkind != RELKIND_RELATION &&
817 [ + + ]: 102 : cstate->rel->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
818 [ + + ]: 79 : cstate->rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE &&
819 [ + + ]: 12 : !(cstate->rel->trigdesc &&
820 [ - + ]: 8 : cstate->rel->trigdesc->trig_insert_instead_row))
821 : : {
822 [ + - ]: 4 : if (cstate->rel->rd_rel->relkind == RELKIND_VIEW)
823 [ + - ]: 4 : ereport(ERROR,
824 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
825 : : errmsg("cannot copy to view \"%s\"",
826 : : RelationGetRelationName(cstate->rel)),
827 : : errhint("To enable copying to a view, provide an INSTEAD OF INSERT trigger.")));
828 [ # # ]: 0 : else if (cstate->rel->rd_rel->relkind == RELKIND_MATVIEW)
829 [ # # ]: 0 : ereport(ERROR,
830 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
831 : : errmsg("cannot copy to materialized view \"%s\"",
832 : : RelationGetRelationName(cstate->rel))));
833 [ # # ]: 0 : else if (cstate->rel->rd_rel->relkind == RELKIND_SEQUENCE)
834 [ # # ]: 0 : ereport(ERROR,
835 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
836 : : errmsg("cannot copy to sequence \"%s\"",
837 : : RelationGetRelationName(cstate->rel))));
838 : : else
839 [ # # ]: 0 : ereport(ERROR,
840 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
841 : : errmsg("cannot copy to non-table relation \"%s\"",
842 : : RelationGetRelationName(cstate->rel))));
843 : : }
844 : :
845 : : /*
846 : : * If the target file is new-in-transaction, we assume that checking FSM
847 : : * for free space is a waste of time. This could possibly be wrong, but
848 : : * it's unlikely.
849 : : */
850 [ + + + - : 1142 : if (RELKIND_HAS_STORAGE(cstate->rel->rd_rel->relkind) &&
+ - + - -
+ ]
851 [ + + ]: 1044 : (cstate->rel->rd_createSubid != InvalidSubTransactionId ||
852 [ + + ]: 1036 : cstate->rel->rd_firstRelfilelocatorSubid != InvalidSubTransactionId))
853 : 53 : ti_options |= TABLE_INSERT_SKIP_FSM;
854 : :
855 : : /*
856 : : * Optimize if new relation storage was created in this subxact or one of
857 : : * its committed children and we won't see those rows later as part of an
858 : : * earlier scan or command. The subxact test ensures that if this subxact
859 : : * aborts then the frozen rows won't be visible after xact cleanup. Note
860 : : * that the stronger test of exactly which subtransaction created it is
861 : : * crucial for correctness of this optimization. The test for an earlier
862 : : * scan or command tolerates false negatives. FREEZE causes other sessions
863 : : * to see rows they would not see under MVCC, and a false negative merely
864 : : * spreads that anomaly to the current session.
865 : : */
866 [ + + ]: 1142 : if (cstate->opts.freeze)
867 : : {
868 : : /*
869 : : * We currently disallow COPY FREEZE on partitioned tables. The
870 : : * reason for this is that we've simply not yet opened the partitions
871 : : * to determine if the optimization can be applied to them. We could
872 : : * go and open them all here, but doing so may be quite a costly
873 : : * overhead for small copies. In any case, we may just end up routing
874 : : * tuples to a small number of partitions. It seems better just to
875 : : * raise an ERROR for partitioned tables.
876 : : */
877 [ + + ]: 45 : if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
878 : : {
879 [ + - ]: 4 : ereport(ERROR,
880 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
881 : : errmsg("cannot perform COPY FREEZE on a partitioned table")));
882 : : }
883 : :
884 : : /* There's currently no support for COPY FREEZE on foreign tables. */
885 [ + + ]: 41 : if (cstate->rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
886 [ + - ]: 4 : ereport(ERROR,
887 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
888 : : errmsg("cannot perform COPY FREEZE on a foreign table")));
889 : :
890 : : /*
891 : : * Tolerate one registration for the benefit of FirstXactSnapshot.
892 : : * Scan-bearing queries generally create at least two registrations,
893 : : * though relying on that is fragile, as is ignoring ActiveSnapshot.
894 : : * Clear CatalogSnapshot to avoid counting its registration. We'll
895 : : * still detect ongoing catalog scans, each of which separately
896 : : * registers the snapshot it uses.
897 : : */
898 : 37 : InvalidateCatalogSnapshot();
899 [ + - - + ]: 37 : if (!ThereAreNoPriorRegisteredSnapshots() || !ThereAreNoReadyPortals())
900 [ # # ]: 0 : ereport(ERROR,
901 : : (errcode(ERRCODE_INVALID_TRANSACTION_STATE),
902 : : errmsg("cannot perform COPY FREEZE because of prior transaction activity")));
903 : :
904 [ + - + + ]: 74 : if (cstate->rel->rd_createSubid != GetCurrentSubTransactionId() &&
905 : 37 : cstate->rel->rd_newRelfilelocatorSubid != GetCurrentSubTransactionId())
906 [ + - ]: 12 : ereport(ERROR,
907 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
908 : : errmsg("cannot perform COPY FREEZE because the table was not created or truncated in the current subtransaction")));
909 : :
910 : 25 : ti_options |= TABLE_INSERT_FROZEN;
911 : : }
912 : :
913 : : /*
914 : : * We need a ResultRelInfo so we can use the regular executor's
915 : : * index-entry-making machinery. (There used to be a huge amount of code
916 : : * here that basically duplicated execUtils.c ...)
917 : : */
918 : 1122 : ExecInitRangeTable(estate, cstate->range_table, cstate->rteperminfos,
919 : : bms_make_singleton(1));
920 : 1122 : resultRelInfo = target_resultRelInfo = makeNode(ResultRelInfo);
921 : 1122 : ExecInitResultRelation(estate, resultRelInfo, 1);
922 : :
923 : : /* Verify the named relation is a valid target for INSERT */
924 : 1122 : CheckValidResultRel(resultRelInfo, CMD_INSERT, ONCONFLICT_NONE, NIL, NULL);
925 : :
926 : 1121 : ExecOpenIndices(resultRelInfo, false);
927 : :
928 : : /*
929 : : * Set up a ModifyTableState so we can let FDW(s) init themselves for
930 : : * foreign-table result relation(s).
931 : : */
932 : 1121 : mtstate = makeNode(ModifyTableState);
933 : 1121 : mtstate->ps.plan = NULL;
934 : 1121 : mtstate->ps.state = estate;
935 : 1121 : mtstate->operation = CMD_INSERT;
936 : 1121 : mtstate->mt_nrels = 1;
937 : 1121 : mtstate->resultRelInfo = resultRelInfo;
938 : 1121 : mtstate->rootResultRelInfo = resultRelInfo;
939 : :
940 [ + + ]: 1121 : if (resultRelInfo->ri_FdwRoutine != NULL &&
941 [ + - ]: 18 : resultRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL)
942 : 18 : resultRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate,
943 : : resultRelInfo);
944 : :
945 : : /*
946 : : * Also, if the named relation is a foreign table, determine if the FDW
947 : : * supports batch insert and determine the batch size (a FDW may support
948 : : * batching, but it may be disabled for the server/table).
949 : : *
950 : : * If the FDW does not support batching, we set the batch size to 1.
951 : : */
952 [ + + ]: 1121 : if (resultRelInfo->ri_FdwRoutine != NULL &&
953 [ + - ]: 18 : resultRelInfo->ri_FdwRoutine->GetForeignModifyBatchSize &&
954 [ + - ]: 18 : resultRelInfo->ri_FdwRoutine->ExecForeignBatchInsert)
955 : 18 : resultRelInfo->ri_BatchSize =
956 : 18 : resultRelInfo->ri_FdwRoutine->GetForeignModifyBatchSize(resultRelInfo);
957 : : else
958 : 1103 : resultRelInfo->ri_BatchSize = 1;
959 : :
960 : : Assert(resultRelInfo->ri_BatchSize >= 1);
961 : :
962 : : /* Prepare to catch AFTER triggers. */
963 : 1121 : AfterTriggerBeginQuery();
964 : :
965 : : /*
966 : : * If there are any triggers with transition tables on the named relation,
967 : : * we need to be prepared to capture transition tuples.
968 : : *
969 : : * Because partition tuple routing would like to know about whether
970 : : * transition capture is active, we also set it in mtstate, which is
971 : : * passed to ExecFindPartition() below.
972 : : */
973 : 1121 : cstate->transition_capture = mtstate->mt_transition_capture =
974 : 1121 : MakeTransitionCaptureState(cstate->rel->trigdesc,
975 : 1121 : RelationGetRelid(cstate->rel),
976 : : CMD_INSERT);
977 : :
978 : : /*
979 : : * If the named relation is a partitioned table, initialize state for
980 : : * CopyFrom tuple routing.
981 : : */
982 [ + + ]: 1121 : if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
983 : 63 : proute = ExecSetupPartitionTupleRouting(estate, cstate->rel);
984 : :
985 [ + + ]: 1121 : if (cstate->whereClause)
986 : 12 : cstate->qualexpr = ExecInitQual(castNode(List, cstate->whereClause),
987 : : &mtstate->ps);
988 : :
989 : : /*
990 : : * It's generally more efficient to prepare a bunch of tuples for
991 : : * insertion, and insert them in one
992 : : * table_multi_insert()/ExecForeignBatchInsert() call, than call
993 : : * table_tuple_insert()/ExecForeignInsert() separately for every tuple.
994 : : * However, there are a number of reasons why we might not be able to do
995 : : * this. These are explained below.
996 : : */
997 [ + + ]: 1121 : if (resultRelInfo->ri_TrigDesc != NULL &&
998 [ + + ]: 121 : (resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
999 [ + + ]: 61 : resultRelInfo->ri_TrigDesc->trig_insert_instead_row))
1000 : : {
1001 : : /*
1002 : : * Can't support multi-inserts when there are any BEFORE/INSTEAD OF
1003 : : * triggers on the table. Such triggers might query the table we're
1004 : : * inserting into and act differently if the tuples that have already
1005 : : * been processed and prepared for insertion are not there.
1006 : : */
1007 : 68 : insertMethod = CIM_SINGLE;
1008 : : }
1009 [ + + ]: 1053 : else if (resultRelInfo->ri_FdwRoutine != NULL &&
1010 [ + + ]: 14 : resultRelInfo->ri_BatchSize == 1)
1011 : : {
1012 : : /*
1013 : : * Can't support multi-inserts to a foreign table if the FDW does not
1014 : : * support batching, or it's disabled for the server or foreign table.
1015 : : */
1016 : 9 : insertMethod = CIM_SINGLE;
1017 : : }
1018 [ + + + + ]: 1044 : else if (proute != NULL && resultRelInfo->ri_TrigDesc != NULL &&
1019 [ + + ]: 20 : resultRelInfo->ri_TrigDesc->trig_insert_new_table)
1020 : : {
1021 : : /*
1022 : : * For partitioned tables we can't support multi-inserts when there
1023 : : * are any statement level insert triggers. It might be possible to
1024 : : * allow partitioned tables with such triggers in the future, but for
1025 : : * now, CopyMultiInsertInfoFlush expects that any after row insert and
1026 : : * statement level insert triggers are on the same relation.
1027 : : */
1028 : 14 : insertMethod = CIM_SINGLE;
1029 : : }
1030 [ + + ]: 1030 : else if (cstate->volatile_defexprs)
1031 : : {
1032 : : /*
1033 : : * Can't support multi-inserts if there are any volatile default
1034 : : * expressions in the table. Similarly to the trigger case above,
1035 : : * such expressions may query the table we're inserting into.
1036 : : *
1037 : : * Note: It does not matter if any partitions have any volatile
1038 : : * default expressions as we use the defaults from the target of the
1039 : : * COPY command.
1040 : : */
1041 : 4 : insertMethod = CIM_SINGLE;
1042 : : }
1043 [ - + ]: 1026 : else if (contain_volatile_functions(cstate->whereClause))
1044 : : {
1045 : : /*
1046 : : * Can't support multi-inserts if there are any volatile function
1047 : : * expressions in WHERE clause. Similarly to the trigger case above,
1048 : : * such expressions may query the table we're inserting into.
1049 : : *
1050 : : * Note: the whereClause was already preprocessed in DoCopy(), so it's
1051 : : * okay to use contain_volatile_functions() directly.
1052 : : */
1053 : 0 : insertMethod = CIM_SINGLE;
1054 : : }
1055 : : else
1056 : : {
1057 : : /*
1058 : : * For partitioned tables, we may still be able to perform bulk
1059 : : * inserts. However, the possibility of this depends on which types
1060 : : * of triggers exist on the partition. We must disable bulk inserts
1061 : : * if the partition is a foreign table that can't use batching or it
1062 : : * has any before row insert or insert instead triggers (same as we
1063 : : * checked above for the parent table). Since the partition's
1064 : : * resultRelInfos are initialized only when we actually need to insert
1065 : : * the first tuple into them, we must have the intermediate insert
1066 : : * method of CIM_MULTI_CONDITIONAL to flag that we must later
1067 : : * determine if we can use bulk-inserts for the partition being
1068 : : * inserted into.
1069 : : */
1070 [ + + ]: 1026 : if (proute)
1071 : 45 : insertMethod = CIM_MULTI_CONDITIONAL;
1072 : : else
1073 : 981 : insertMethod = CIM_MULTI;
1074 : :
1075 : 1026 : CopyMultiInsertInfoInit(&multiInsertInfo, resultRelInfo, cstate,
1076 : : estate, mycid, ti_options);
1077 : : }
1078 : :
1079 : : /*
1080 : : * If not using batch mode (which allocates slots as needed) set up a
1081 : : * tuple slot too. When inserting into a partitioned table, we also need
1082 : : * one, even if we might batch insert, to read the tuple in the root
1083 : : * partition's form.
1084 : : */
1085 [ + + + + ]: 1121 : if (insertMethod == CIM_SINGLE || insertMethod == CIM_MULTI_CONDITIONAL)
1086 : : {
1087 : 140 : singleslot = table_slot_create(resultRelInfo->ri_RelationDesc,
1088 : : &estate->es_tupleTable);
1089 : 140 : bistate = GetBulkInsertState();
1090 : : }
1091 : :
1092 [ + + ]: 1242 : has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
1093 [ + + ]: 121 : resultRelInfo->ri_TrigDesc->trig_insert_before_row);
1094 : :
1095 [ + + ]: 1242 : has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
1096 [ + + ]: 121 : resultRelInfo->ri_TrigDesc->trig_insert_instead_row);
1097 : :
1098 : : /*
1099 : : * Check BEFORE STATEMENT insertion triggers. It's debatable whether we
1100 : : * should do this for COPY, since it's not really an "INSERT" statement as
1101 : : * such. However, executing these triggers maintains consistency with the
1102 : : * EACH ROW triggers that we already fire on COPY.
1103 : : */
1104 : 1121 : ExecBSInsertTriggers(estate, resultRelInfo);
1105 : :
1106 [ + + ]: 1121 : econtext = GetPerTupleExprContext(estate);
1107 : :
1108 : : /* Set up callback to identify error line number */
1109 : 1121 : errcallback.callback = CopyFromErrorCallback;
1110 : 1121 : errcallback.arg = cstate;
1111 : 1121 : errcallback.previous = error_context_stack;
1112 : 1121 : error_context_stack = &errcallback;
1113 : :
1114 : : for (;;)
1115 : 758502 : {
1116 : : TupleTableSlot *myslot;
1117 : : bool skip_tuple;
1118 : :
1119 [ - + ]: 759623 : CHECK_FOR_INTERRUPTS();
1120 : :
1121 : : /*
1122 : : * Reset the per-tuple exprcontext. We do this after every tuple, to
1123 : : * clean-up after expression evaluations etc.
1124 : : */
1125 [ + - ]: 759623 : ResetPerTupleExprContext(estate);
1126 : :
1127 : : /* select slot to (initially) load row into */
1128 [ + + + + ]: 759623 : if (insertMethod == CIM_SINGLE || proute)
1129 : : {
1130 : 149557 : myslot = singleslot;
1131 : 149557 : Assert(myslot != NULL);
1132 : : }
1133 : : else
1134 : : {
1135 : : Assert(resultRelInfo == target_resultRelInfo);
1136 : : Assert(insertMethod == CIM_MULTI);
1137 : :
1138 : 610066 : myslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
1139 : : resultRelInfo);
1140 : : }
1141 : :
1142 : : /*
1143 : : * Switch to per-tuple context before calling NextCopyFrom, which does
1144 : : * evaluate default expressions etc. and requires per-tuple context.
1145 : : */
1146 [ + - ]: 759623 : MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
1147 : :
1148 : 759623 : ExecClearTuple(myslot);
1149 : :
1150 : : /* Directly store the values/nulls array in the slot */
1151 [ + + ]: 759623 : if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
1152 : 966 : break;
1153 : :
1154 [ + + ]: 758530 : if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE &&
1155 [ + + ]: 108 : cstate->escontext->error_occurred)
1156 : : {
1157 : : /*
1158 : : * Soft error occurred, skip this tuple and just make
1159 : : * ErrorSaveContext ready for the next NextCopyFrom. Since we
1160 : : * don't set details_wanted and error_data is not to be filled,
1161 : : * just resetting error_occurred is enough.
1162 : : */
1163 : 72 : cstate->escontext->error_occurred = false;
1164 : :
1165 : : /* Report that this tuple was skipped by the ON_ERROR clause */
1166 : 72 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_SKIPPED,
1167 : 72 : cstate->num_errors);
1168 : :
1169 [ + + ]: 72 : if (cstate->opts.reject_limit > 0 &&
1170 [ + + ]: 32 : cstate->num_errors > cstate->opts.reject_limit)
1171 [ + - ]: 4 : ereport(ERROR,
1172 : : (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1173 : : errmsg("skipped more than REJECT_LIMIT (%" PRId64 ") rows due to data type incompatibility",
1174 : : cstate->opts.reject_limit)));
1175 : :
1176 : : /* Repeat NextCopyFrom() until no soft error occurs */
1177 : 68 : continue;
1178 : : }
1179 : :
1180 : 758458 : ExecStoreVirtualTuple(myslot);
1181 : :
1182 : : /*
1183 : : * Constraints and where clause might reference the tableoid column,
1184 : : * so (re-)initialize tts_tableOid before evaluating them.
1185 : : */
1186 : 758458 : myslot->tts_tableOid = RelationGetRelid(target_resultRelInfo->ri_RelationDesc);
1187 : :
1188 : : /* Triggers and stuff need to be invoked in query context. */
1189 : 758458 : MemoryContextSwitchTo(oldcontext);
1190 : :
1191 [ + + ]: 758458 : if (cstate->whereClause)
1192 : : {
1193 : 44 : econtext->ecxt_scantuple = myslot;
1194 : : /* Skip items that don't match COPY's WHERE clause */
1195 [ + + ]: 44 : if (!ExecQual(cstate->qualexpr, econtext))
1196 : : {
1197 : : /*
1198 : : * Report that this tuple was filtered out by the WHERE
1199 : : * clause.
1200 : : */
1201 : 24 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED,
1202 : : ++excluded);
1203 : 24 : continue;
1204 : : }
1205 : : }
1206 : :
1207 : : /* Determine the partition to insert the tuple into */
1208 [ + + ]: 758434 : if (proute)
1209 : : {
1210 : : TupleConversionMap *map;
1211 : :
1212 : : /*
1213 : : * Attempt to find a partition suitable for this tuple.
1214 : : * ExecFindPartition() will raise an error if none can be found or
1215 : : * if the found partition is not suitable for INSERTs.
1216 : : */
1217 : 149252 : resultRelInfo = ExecFindPartition(mtstate, target_resultRelInfo,
1218 : : proute, myslot, estate);
1219 : :
1220 [ + + ]: 149251 : if (prevResultRelInfo != resultRelInfo)
1221 : : {
1222 : : /* Determine which triggers exist on this partition */
1223 [ + + ]: 90815 : has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
1224 [ + + ]: 36 : resultRelInfo->ri_TrigDesc->trig_insert_before_row);
1225 : :
1226 [ + + ]: 90815 : has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
1227 [ - + ]: 36 : resultRelInfo->ri_TrigDesc->trig_insert_instead_row);
1228 : :
1229 : : /*
1230 : : * Disable multi-inserts when the partition has BEFORE/INSTEAD
1231 : : * OF triggers, or if the partition is a foreign table that
1232 : : * can't use batching.
1233 : : */
1234 : 141520 : leafpart_use_multi_insert = insertMethod == CIM_MULTI_CONDITIONAL &&
1235 [ + + ]: 50741 : !has_before_insert_row_trig &&
1236 [ + + + - ]: 192245 : !has_instead_insert_row_trig &&
1237 [ + + ]: 50725 : (resultRelInfo->ri_FdwRoutine == NULL ||
1238 [ + + ]: 6 : resultRelInfo->ri_BatchSize > 1);
1239 : :
1240 : : /* Set the multi-insert buffer to use for this partition. */
1241 [ + + ]: 90779 : if (leafpart_use_multi_insert)
1242 : : {
1243 [ + + ]: 50723 : if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL)
1244 : 49 : CopyMultiInsertInfoSetupBuffer(&multiInsertInfo,
1245 : : resultRelInfo);
1246 : : }
1247 [ + + ]: 40056 : else if (insertMethod == CIM_MULTI_CONDITIONAL &&
1248 [ - + ]: 18 : !CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
1249 : : {
1250 : : /*
1251 : : * Flush pending inserts if this partition can't use
1252 : : * batching, so rows are visible to triggers etc.
1253 : : */
1254 : 0 : CopyMultiInsertInfoFlush(&multiInsertInfo,
1255 : : resultRelInfo,
1256 : : &processed);
1257 : : }
1258 : :
1259 [ + - ]: 90779 : if (bistate != NULL)
1260 : 90779 : ReleaseBulkInsertStatePin(bistate);
1261 : 90779 : prevResultRelInfo = resultRelInfo;
1262 : : }
1263 : :
1264 : : /*
1265 : : * If we're capturing transition tuples, we might need to convert
1266 : : * from the partition rowtype to root rowtype. But if there are no
1267 : : * BEFORE triggers on the partition that could change the tuple,
1268 : : * we can just remember the original unconverted tuple to avoid a
1269 : : * needless round trip conversion.
1270 : : */
1271 [ + + ]: 149251 : if (cstate->transition_capture != NULL)
1272 : 38 : cstate->transition_capture->tcs_original_insert_tuple =
1273 [ + + ]: 38 : !has_before_insert_row_trig ? myslot : NULL;
1274 : :
1275 : : /*
1276 : : * We might need to convert from the root rowtype to the partition
1277 : : * rowtype.
1278 : : */
1279 : 149251 : map = ExecGetRootToChildMap(resultRelInfo, estate);
1280 [ + + + + ]: 149251 : if (insertMethod == CIM_SINGLE || !leafpart_use_multi_insert)
1281 : : {
1282 : : /* non batch insert */
1283 [ + + ]: 40092 : if (map != NULL)
1284 : : {
1285 : : TupleTableSlot *new_slot;
1286 : :
1287 : 73 : new_slot = resultRelInfo->ri_PartitionTupleSlot;
1288 : 73 : myslot = execute_attr_map_slot(map->attrMap, myslot, new_slot);
1289 : : }
1290 : : }
1291 : : else
1292 : : {
1293 : : /*
1294 : : * Prepare to queue up tuple for later batch insert into
1295 : : * current partition.
1296 : : */
1297 : : TupleTableSlot *batchslot;
1298 : :
1299 : : /* no other path available for partitioned table */
1300 : : Assert(insertMethod == CIM_MULTI_CONDITIONAL);
1301 : :
1302 : 109159 : batchslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
1303 : : resultRelInfo);
1304 : :
1305 [ + + ]: 109159 : if (map != NULL)
1306 : 8132 : myslot = execute_attr_map_slot(map->attrMap, myslot,
1307 : : batchslot);
1308 : : else
1309 : : {
1310 : : /*
1311 : : * This looks more expensive than it is (Believe me, I
1312 : : * optimized it away. Twice.). The input is in virtual
1313 : : * form, and we'll materialize the slot below - for most
1314 : : * slot types the copy performs the work materialization
1315 : : * would later require anyway.
1316 : : */
1317 : 101027 : ExecCopySlot(batchslot, myslot);
1318 : 101027 : myslot = batchslot;
1319 : : }
1320 : : }
1321 : :
1322 : : /* ensure that triggers etc see the right relation */
1323 : 149251 : myslot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
1324 : : }
1325 : :
1326 : 758433 : skip_tuple = false;
1327 : :
1328 : : /* BEFORE ROW INSERT Triggers */
1329 [ + + ]: 758433 : if (has_before_insert_row_trig)
1330 : : {
1331 [ + + ]: 180 : if (!ExecBRInsertTriggers(estate, resultRelInfo, myslot))
1332 : 10 : skip_tuple = true; /* "do nothing" */
1333 : : }
1334 : :
1335 [ + + ]: 758433 : if (!skip_tuple)
1336 : : {
1337 : : /*
1338 : : * If there is an INSTEAD OF INSERT ROW trigger, let it handle the
1339 : : * tuple. Otherwise, proceed with inserting the tuple into the
1340 : : * table or foreign table.
1341 : : */
1342 [ + + ]: 758423 : if (has_instead_insert_row_trig)
1343 : : {
1344 : 8 : ExecIRInsertTriggers(estate, resultRelInfo, myslot);
1345 : : }
1346 : : else
1347 : : {
1348 : : /* Compute stored generated columns */
1349 [ + + ]: 758415 : if (resultRelInfo->ri_RelationDesc->rd_att->constr &&
1350 [ + + ]: 241113 : resultRelInfo->ri_RelationDesc->rd_att->constr->has_generated_stored)
1351 : 21 : ExecComputeStoredGenerated(resultRelInfo, estate, myslot,
1352 : : CMD_INSERT);
1353 : :
1354 : : /*
1355 : : * If the target is a plain table, check the constraints of
1356 : : * the tuple.
1357 : : */
1358 [ + + ]: 758415 : if (resultRelInfo->ri_FdwRoutine == NULL &&
1359 [ + + ]: 758369 : resultRelInfo->ri_RelationDesc->rd_att->constr)
1360 : 241095 : ExecConstraints(resultRelInfo, myslot, estate);
1361 : :
1362 : : /*
1363 : : * Also check the tuple against the partition constraint, if
1364 : : * there is one; except that if we got here via tuple-routing,
1365 : : * we don't need to if there's no BR trigger defined on the
1366 : : * partition.
1367 : : */
1368 [ + + + + ]: 758395 : if (resultRelInfo->ri_RelationDesc->rd_rel->relispartition &&
1369 [ + + ]: 149243 : (proute == NULL || has_before_insert_row_trig))
1370 : 1167 : ExecPartitionCheck(resultRelInfo, myslot, estate, true);
1371 : :
1372 : : /* Store the slot in the multi-insert buffer, when enabled. */
1373 [ + + + + ]: 758395 : if (insertMethod == CIM_MULTI || leafpart_use_multi_insert)
1374 : : {
1375 : : /*
1376 : : * The slot previously might point into the per-tuple
1377 : : * context. For batching it needs to be longer lived.
1378 : : */
1379 : 718172 : ExecMaterializeSlot(myslot);
1380 : :
1381 : : /* Add this tuple to the tuple buffer */
1382 : 718172 : CopyMultiInsertInfoStore(&multiInsertInfo,
1383 : : resultRelInfo, myslot,
1384 : : cstate->line_buf.len,
1385 : : cstate->cur_lineno);
1386 : :
1387 : : /*
1388 : : * If enough inserts have queued up, then flush all
1389 : : * buffers out to their tables.
1390 : : */
1391 [ + + ]: 718172 : if (CopyMultiInsertInfoIsFull(&multiInsertInfo))
1392 : 656 : CopyMultiInsertInfoFlush(&multiInsertInfo,
1393 : : resultRelInfo,
1394 : : &processed);
1395 : :
1396 : : /*
1397 : : * We delay updating the row counter and progress of the
1398 : : * COPY command until after writing the tuples stored in
1399 : : * the buffer out to the table, as in single insert mode.
1400 : : * See CopyMultiInsertBufferFlush().
1401 : : */
1402 : 718172 : continue; /* next tuple please */
1403 : : }
1404 : : else
1405 : : {
1406 : 40223 : List *recheckIndexes = NIL;
1407 : :
1408 : : /* OK, store the tuple */
1409 [ + + ]: 40223 : if (resultRelInfo->ri_FdwRoutine != NULL)
1410 : : {
1411 : 27 : myslot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
1412 : : resultRelInfo,
1413 : : myslot,
1414 : : NULL);
1415 : :
1416 [ + + ]: 26 : if (myslot == NULL) /* "do nothing" */
1417 : 2 : continue; /* next tuple please */
1418 : :
1419 : : /*
1420 : : * AFTER ROW Triggers might reference the tableoid
1421 : : * column, so (re-)initialize tts_tableOid before
1422 : : * evaluating them.
1423 : : */
1424 : 24 : myslot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
1425 : : }
1426 : : else
1427 : : {
1428 : : /* OK, store the tuple and create index entries for it */
1429 : 40196 : table_tuple_insert(resultRelInfo->ri_RelationDesc,
1430 : : myslot, mycid, ti_options, bistate);
1431 : :
1432 [ - + ]: 40196 : if (resultRelInfo->ri_NumIndices > 0)
1433 : 0 : recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
1434 : : estate, 0,
1435 : : myslot, NIL,
1436 : : NULL);
1437 : : }
1438 : :
1439 : : /* AFTER ROW INSERT Triggers */
1440 : 40220 : ExecARInsertTriggers(estate, resultRelInfo, myslot,
1441 : : recheckIndexes, cstate->transition_capture);
1442 : :
1443 : 40218 : list_free(recheckIndexes);
1444 : : }
1445 : : }
1446 : :
1447 : : /*
1448 : : * We count only tuples not suppressed by a BEFORE INSERT trigger
1449 : : * or FDW; this is the same definition used by nodeModifyTable.c
1450 : : * for counting tuples inserted by an INSERT command. Update
1451 : : * progress of the COPY command as well.
1452 : : */
1453 : 40226 : pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
1454 : : ++processed);
1455 : : }
1456 : : }
1457 : :
1458 : : /* Flush any remaining buffered tuples */
1459 [ + + ]: 966 : if (insertMethod != CIM_SINGLE)
1460 : : {
1461 [ + + ]: 890 : if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
1462 : 711 : CopyMultiInsertInfoFlush(&multiInsertInfo, NULL, &processed);
1463 : : }
1464 : :
1465 : : /* Done, clean up */
1466 : 956 : error_context_stack = errcallback.previous;
1467 : :
1468 [ + + ]: 956 : if (cstate->num_errors > 0 &&
1469 [ + + ]: 24 : cstate->opts.log_verbosity >= COPY_LOG_VERBOSITY_DEFAULT)
1470 : : {
1471 [ + + ]: 20 : if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE)
1472 [ + - ]: 16 : ereport(NOTICE,
1473 : : errmsg_plural("%" PRIu64 " row was skipped due to data type incompatibility",
1474 : : "%" PRIu64 " rows were skipped due to data type incompatibility",
1475 : : cstate->num_errors,
1476 : : cstate->num_errors));
1477 [ + - ]: 4 : else if (cstate->opts.on_error == COPY_ON_ERROR_SET_NULL)
1478 [ + - ]: 4 : ereport(NOTICE,
1479 : : errmsg_plural("in %" PRIu64 " row, columns were set to null due to data type incompatibility",
1480 : : "in %" PRIu64 " rows, columns were set to null due to data type incompatibility",
1481 : : cstate->num_errors,
1482 : : cstate->num_errors));
1483 : : }
1484 : :
1485 [ + + ]: 956 : if (bistate != NULL)
1486 : 120 : FreeBulkInsertState(bistate);
1487 : :
1488 : 956 : MemoryContextSwitchTo(oldcontext);
1489 : :
1490 : : /* Execute AFTER STATEMENT insertion triggers */
1491 : 956 : ExecASInsertTriggers(estate, target_resultRelInfo, cstate->transition_capture);
1492 : :
1493 : : /* Handle queued AFTER triggers */
1494 : 956 : AfterTriggerEndQuery(estate);
1495 : :
1496 : 956 : ExecResetTupleTable(estate->es_tupleTable, false);
1497 : :
1498 : : /* Allow the FDW to shut down */
1499 [ + + ]: 956 : if (target_resultRelInfo->ri_FdwRoutine != NULL &&
1500 [ + - ]: 16 : target_resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL)
1501 : 16 : target_resultRelInfo->ri_FdwRoutine->EndForeignInsert(estate,
1502 : : target_resultRelInfo);
1503 : :
1504 : : /* Tear down the multi-insert buffer data */
1505 [ + + ]: 956 : if (insertMethod != CIM_SINGLE)
1506 : 880 : CopyMultiInsertInfoCleanup(&multiInsertInfo);
1507 : :
1508 : : /* Close all the partitioned tables, leaf partitions, and their indices */
1509 [ + + ]: 956 : if (proute)
1510 : 60 : ExecCleanupTupleRouting(mtstate, proute);
1511 : :
1512 : : /* Close the result relations, including any trigger target relations */
1513 : 956 : ExecCloseResultRelations(estate);
1514 : 956 : ExecCloseRangeTableRelations(estate);
1515 : :
1516 : 956 : FreeExecutorState(estate);
1517 : :
1518 : 956 : return processed;
1519 : : }
1520 : :
1521 : : /*
1522 : : * Setup to read tuples from a file for COPY FROM.
1523 : : *
1524 : : * 'rel': Used as a template for the tuples
1525 : : * 'whereClause': WHERE clause from the COPY FROM command
1526 : : * 'filename': Name of server-local file to read, NULL for STDIN
1527 : : * 'is_program': true if 'filename' is program to execute
1528 : : * 'data_source_cb': callback that provides the input data
1529 : : * 'attnamelist': List of char *, columns to include. NIL selects all cols.
1530 : : * 'options': List of DefElem. See copy_opt_item in gram.y for selections.
1531 : : *
1532 : : * Returns a CopyFromState, to be passed to NextCopyFrom and related functions.
1533 : : */
1534 : : CopyFromState
1535 : 1382 : BeginCopyFrom(ParseState *pstate,
1536 : : Relation rel,
1537 : : Node *whereClause,
1538 : : const char *filename,
1539 : : bool is_program,
1540 : : copy_data_source_cb data_source_cb,
1541 : : List *attnamelist,
1542 : : List *options)
1543 : : {
1544 : : CopyFromState cstate;
1545 : 1382 : bool pipe = (filename == NULL);
1546 : : TupleDesc tupDesc;
1547 : : AttrNumber num_phys_attrs,
1548 : : num_defaults;
1549 : : FmgrInfo *in_functions;
1550 : : Oid *typioparams;
1551 : : int *defmap;
1552 : : ExprState **defexprs;
1553 : : MemoryContext oldcontext;
1554 : : bool volatile_defexprs;
1555 : 1382 : const int progress_cols[] = {
1556 : : PROGRESS_COPY_COMMAND,
1557 : : PROGRESS_COPY_TYPE,
1558 : : PROGRESS_COPY_BYTES_TOTAL
1559 : : };
1560 : 1382 : int64 progress_vals[] = {
1561 : : PROGRESS_COPY_COMMAND_FROM,
1562 : : 0,
1563 : : 0
1564 : : };
1565 : :
1566 : : /* Allocate workspace and zero all fields */
1567 : 1382 : cstate = palloc0_object(CopyFromStateData);
1568 : :
1569 : : /*
1570 : : * We allocate everything used by a cstate in a new memory context. This
1571 : : * avoids memory leaks during repeated use of COPY in a query.
1572 : : */
1573 : 1382 : cstate->copycontext = AllocSetContextCreate(CurrentMemoryContext,
1574 : : "COPY",
1575 : : ALLOCSET_DEFAULT_SIZES);
1576 : :
1577 : 1382 : oldcontext = MemoryContextSwitchTo(cstate->copycontext);
1578 : :
1579 : : /* Extract options from the statement node tree */
1580 : 1382 : ProcessCopyOptions(pstate, &cstate->opts, true /* is_from */ , options);
1581 : :
1582 : : /* Set the format routine */
1583 : 1201 : cstate->routine = CopyFromGetRoutine(&cstate->opts);
1584 : :
1585 : : /* Process the target relation */
1586 : 1201 : cstate->rel = rel;
1587 : :
1588 : 1201 : tupDesc = RelationGetDescr(cstate->rel);
1589 : :
1590 : : /* process common options or initialization */
1591 : :
1592 : : /* Generate or convert list of attributes to process */
1593 : 1201 : cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
1594 : :
1595 : 1201 : num_phys_attrs = tupDesc->natts;
1596 : :
1597 : : /* Convert FORCE_NOT_NULL name list to per-column flags, check validity */
1598 : 1201 : cstate->opts.force_notnull_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
1599 [ + + ]: 1201 : if (cstate->opts.force_notnull_all)
1600 [ + - - + : 8 : MemSet(cstate->opts.force_notnull_flags, true, num_phys_attrs * sizeof(bool));
- - - - -
- ]
1601 [ + + ]: 1193 : else if (cstate->opts.force_notnull)
1602 : : {
1603 : : List *attnums;
1604 : : ListCell *cur;
1605 : :
1606 : 17 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_notnull);
1607 : :
1608 [ + - + + : 34 : foreach(cur, attnums)
+ + ]
1609 : : {
1610 : 21 : int attnum = lfirst_int(cur);
1611 : 21 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1612 : :
1613 [ + + ]: 21 : if (!list_member_int(cstate->attnumlist, attnum))
1614 [ + - ]: 4 : ereport(ERROR,
1615 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1616 : : /*- translator: first %s is the name of a COPY option, e.g. FORCE_NOT_NULL */
1617 : : errmsg("%s column \"%s\" not referenced by COPY",
1618 : : "FORCE_NOT_NULL", NameStr(attr->attname))));
1619 : 17 : cstate->opts.force_notnull_flags[attnum - 1] = true;
1620 : : }
1621 : : }
1622 : :
1623 : : /* Set up soft error handler for ON_ERROR */
1624 [ + + ]: 1197 : if (cstate->opts.on_error != COPY_ON_ERROR_STOP)
1625 : : {
1626 : 70 : cstate->escontext = makeNode(ErrorSaveContext);
1627 : 70 : cstate->escontext->type = T_ErrorSaveContext;
1628 : 70 : cstate->escontext->error_occurred = false;
1629 : :
1630 [ + + ]: 70 : if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE ||
1631 [ + - ]: 29 : cstate->opts.on_error == COPY_ON_ERROR_SET_NULL)
1632 : 70 : cstate->escontext->details_wanted = false;
1633 : : }
1634 : : else
1635 : 1127 : cstate->escontext = NULL;
1636 : :
1637 [ + + ]: 1197 : if (cstate->opts.on_error == COPY_ON_ERROR_SET_NULL)
1638 : : {
1639 : : /*
1640 : : * When data type conversion fails and ON_ERROR is SET_NULL, we need
1641 : : * ensure that the input column allow null values. ExecConstraints()
1642 : : * will cover most of the cases, but it does not verify domain
1643 : : * constraints. Therefore, for constrained domains, the null value
1644 : : * check must be performed during the initial string-to-datum
1645 : : * conversion (see CopyFromTextLikeOneRow()).
1646 : : */
1647 : 29 : cstate->domain_with_constraint = palloc0_array(bool, num_phys_attrs);
1648 : :
1649 [ + - + + : 140 : foreach_int(attno, cstate->attnumlist)
+ + ]
1650 : : {
1651 : 82 : Form_pg_attribute att = TupleDescAttr(tupDesc, attno - 1);
1652 : :
1653 : 82 : cstate->domain_with_constraint[attno - 1] = DomainHasConstraints(att->atttypid, NULL);
1654 : : }
1655 : : }
1656 : :
1657 : : /* Convert FORCE_NULL name list to per-column flags, check validity */
1658 : 1197 : cstate->opts.force_null_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
1659 [ + + ]: 1197 : if (cstate->opts.force_null_all)
1660 [ + - - + : 8 : MemSet(cstate->opts.force_null_flags, true, num_phys_attrs * sizeof(bool));
- - - - -
- ]
1661 [ + + ]: 1189 : else if (cstate->opts.force_null)
1662 : : {
1663 : : List *attnums;
1664 : : ListCell *cur;
1665 : :
1666 : 17 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_null);
1667 : :
1668 [ + - + + : 34 : foreach(cur, attnums)
+ + ]
1669 : : {
1670 : 21 : int attnum = lfirst_int(cur);
1671 : 21 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1672 : :
1673 [ + + ]: 21 : if (!list_member_int(cstate->attnumlist, attnum))
1674 [ + - ]: 4 : ereport(ERROR,
1675 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1676 : : /*- translator: first %s is the name of a COPY option, e.g. FORCE_NOT_NULL */
1677 : : errmsg("%s column \"%s\" not referenced by COPY",
1678 : : "FORCE_NULL", NameStr(attr->attname))));
1679 : 17 : cstate->opts.force_null_flags[attnum - 1] = true;
1680 : : }
1681 : : }
1682 : :
1683 : : /* Convert convert_selectively name list to per-column flags */
1684 [ + + ]: 1193 : if (cstate->opts.convert_selectively)
1685 : : {
1686 : : List *attnums;
1687 : : ListCell *cur;
1688 : :
1689 : 3 : cstate->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
1690 : :
1691 : 3 : attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.convert_select);
1692 : :
1693 [ + - + + : 7 : foreach(cur, attnums)
+ + ]
1694 : : {
1695 : 4 : int attnum = lfirst_int(cur);
1696 : 4 : Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
1697 : :
1698 [ - + ]: 4 : if (!list_member_int(cstate->attnumlist, attnum))
1699 [ # # ]: 0 : ereport(ERROR,
1700 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1701 : : errmsg_internal("selected column \"%s\" not referenced by COPY",
1702 : : NameStr(attr->attname))));
1703 : 4 : cstate->convert_select_flags[attnum - 1] = true;
1704 : : }
1705 : : }
1706 : :
1707 : : /* Use client encoding when ENCODING option is not specified. */
1708 [ + + ]: 1193 : if (cstate->opts.file_encoding < 0)
1709 : 1173 : cstate->file_encoding = pg_get_client_encoding();
1710 : : else
1711 : 20 : cstate->file_encoding = cstate->opts.file_encoding;
1712 : :
1713 : : /*
1714 : : * Look up encoding conversion function.
1715 : : */
1716 [ + + ]: 1193 : if (cstate->file_encoding == GetDatabaseEncoding() ||
1717 [ + + - + ]: 52 : cstate->file_encoding == PG_SQL_ASCII ||
1718 : 24 : GetDatabaseEncoding() == PG_SQL_ASCII)
1719 : : {
1720 : 1169 : cstate->need_transcoding = false;
1721 : : }
1722 : : else
1723 : : {
1724 : 24 : cstate->need_transcoding = true;
1725 : 24 : cstate->conversion_proc = FindDefaultConversionProc(cstate->file_encoding,
1726 : : GetDatabaseEncoding());
1727 [ - + ]: 24 : if (!OidIsValid(cstate->conversion_proc))
1728 [ # # ]: 0 : ereport(ERROR,
1729 : : (errcode(ERRCODE_UNDEFINED_FUNCTION),
1730 : : errmsg("default conversion function for encoding \"%s\" to \"%s\" does not exist",
1731 : : pg_encoding_to_char(cstate->file_encoding),
1732 : : pg_encoding_to_char(GetDatabaseEncoding()))));
1733 : : }
1734 : :
1735 : 1193 : cstate->copy_src = COPY_FILE; /* default */
1736 : :
1737 : 1193 : cstate->whereClause = whereClause;
1738 : :
1739 : : /* Initialize state variables */
1740 : 1193 : cstate->eol_type = EOL_UNKNOWN;
1741 : 1193 : cstate->cur_relname = RelationGetRelationName(cstate->rel);
1742 : 1193 : cstate->cur_lineno = 0;
1743 : 1193 : cstate->cur_attname = NULL;
1744 : 1193 : cstate->cur_attval = NULL;
1745 : 1193 : cstate->relname_only = false;
1746 : 1193 : cstate->simd_enabled = true;
1747 : :
1748 : : /*
1749 : : * Allocate buffers for the input pipeline.
1750 : : *
1751 : : * attribute_buf and raw_buf are used in both text and binary modes, but
1752 : : * input_buf and line_buf only in text mode.
1753 : : */
1754 : 1193 : cstate->raw_buf = palloc(RAW_BUF_SIZE + 1);
1755 : 1193 : cstate->raw_buf_index = cstate->raw_buf_len = 0;
1756 : 1193 : cstate->raw_reached_eof = false;
1757 : :
1758 : 1193 : initStringInfo(&cstate->attribute_buf);
1759 : :
1760 : : /* Assign range table and rteperminfos, we'll need them in CopyFrom. */
1761 [ + + ]: 1193 : if (pstate)
1762 : : {
1763 : 1155 : cstate->range_table = pstate->p_rtable;
1764 : 1155 : cstate->rteperminfos = pstate->p_rteperminfos;
1765 : : }
1766 : :
1767 : 1193 : num_defaults = 0;
1768 : 1193 : volatile_defexprs = false;
1769 : :
1770 : : /*
1771 : : * Pick up the required catalog information for each attribute in the
1772 : : * relation, including the input function, the element type (to pass to
1773 : : * the input function), and info about defaults and constraints. (Which
1774 : : * input function we use depends on text/binary format choice.)
1775 : : */
1776 : 1193 : in_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
1777 : 1193 : typioparams = (Oid *) palloc(num_phys_attrs * sizeof(Oid));
1778 : 1193 : defmap = (int *) palloc(num_phys_attrs * sizeof(int));
1779 : 1193 : defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));
1780 : :
1781 [ + + ]: 4758 : for (int attnum = 1; attnum <= num_phys_attrs; attnum++)
1782 : : {
1783 : 3574 : Form_pg_attribute att = TupleDescAttr(tupDesc, attnum - 1);
1784 : :
1785 : : /* We don't need info for dropped attributes */
1786 [ + + ]: 3574 : if (att->attisdropped)
1787 : 82 : continue;
1788 : :
1789 : : /* Fetch the input function and typioparam info */
1790 : 3492 : cstate->routine->CopyFromInFunc(cstate, att->atttypid,
1791 : 3492 : &in_functions[attnum - 1],
1792 : 3492 : &typioparams[attnum - 1]);
1793 : :
1794 : : /* Get default info if available */
1795 : 3491 : defexprs[attnum - 1] = NULL;
1796 : :
1797 : : /*
1798 : : * We only need the default values for columns that do not appear in
1799 : : * the column list, unless the DEFAULT option was given. We never need
1800 : : * default values for generated columns.
1801 : : */
1802 [ + + ]: 3491 : if ((cstate->opts.default_print != NULL ||
1803 [ + + ]: 3409 : !list_member_int(cstate->attnumlist, attnum)) &&
1804 [ + + ]: 425 : !att->attgenerated)
1805 : : {
1806 : 403 : Expr *defexpr = (Expr *) build_column_default(cstate->rel,
1807 : : attnum);
1808 : :
1809 [ + + ]: 403 : if (defexpr != NULL)
1810 : : {
1811 : : /* Run the expression through planner */
1812 : 167 : defexpr = expression_planner(defexpr);
1813 : :
1814 : : /* Initialize executable expression in copycontext */
1815 : 159 : defexprs[attnum - 1] = ExecInitExpr(defexpr, NULL);
1816 : :
1817 : : /* if NOT copied from input */
1818 : : /* use default value if one exists */
1819 [ + + ]: 159 : if (!list_member_int(cstate->attnumlist, attnum))
1820 : : {
1821 : 107 : defmap[num_defaults] = attnum - 1;
1822 : 107 : num_defaults++;
1823 : : }
1824 : :
1825 : : /*
1826 : : * If a default expression looks at the table being loaded,
1827 : : * then it could give the wrong answer when using
1828 : : * multi-insert. Since database access can be dynamic this is
1829 : : * hard to test for exactly, so we use the much wider test of
1830 : : * whether the default expression is volatile. We allow for
1831 : : * the special case of when the default expression is the
1832 : : * nextval() of a sequence which in this specific case is
1833 : : * known to be safe for use with the multi-insert
1834 : : * optimization. Hence we use this special case function
1835 : : * checker rather than the standard check for
1836 : : * contain_volatile_functions(). Note also that we already
1837 : : * ran the expression through expression_planner().
1838 : : */
1839 [ + - ]: 159 : if (!volatile_defexprs)
1840 : 159 : volatile_defexprs = contain_volatile_functions_not_nextval((Node *) defexpr);
1841 : : }
1842 : : }
1843 : : }
1844 : :
1845 : 1184 : cstate->defaults = (bool *) palloc0(tupDesc->natts * sizeof(bool));
1846 : :
1847 : : /* initialize progress */
1848 : 1184 : pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
1849 [ + - ]: 1184 : cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
1850 : 1184 : cstate->bytes_processed = 0;
1851 : :
1852 : : /* We keep those variables in cstate. */
1853 : 1184 : cstate->in_functions = in_functions;
1854 : 1184 : cstate->typioparams = typioparams;
1855 : 1184 : cstate->defmap = defmap;
1856 : 1184 : cstate->defexprs = defexprs;
1857 : 1184 : cstate->volatile_defexprs = volatile_defexprs;
1858 : 1184 : cstate->num_defaults = num_defaults;
1859 : 1184 : cstate->is_program = is_program;
1860 : :
1861 [ + + ]: 1184 : if (data_source_cb)
1862 : : {
1863 : 208 : progress_vals[1] = PROGRESS_COPY_TYPE_CALLBACK;
1864 : 208 : cstate->copy_src = COPY_CALLBACK;
1865 : 208 : cstate->data_source_cb = data_source_cb;
1866 : : }
1867 [ + + ]: 976 : else if (pipe)
1868 : : {
1869 : 689 : progress_vals[1] = PROGRESS_COPY_TYPE_PIPE;
1870 : : Assert(!is_program); /* the grammar does not allow this */
1871 [ + - ]: 689 : if (whereToSendOutput == DestRemote)
1872 : 689 : ReceiveCopyBegin(cstate);
1873 : : else
1874 : 0 : cstate->copy_file = stdin;
1875 : : }
1876 : : else
1877 : : {
1878 : 287 : cstate->filename = pstrdup(filename);
1879 : :
1880 [ - + ]: 287 : if (cstate->is_program)
1881 : : {
1882 : 0 : progress_vals[1] = PROGRESS_COPY_TYPE_PROGRAM;
1883 : 0 : cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_R);
1884 [ # # ]: 0 : if (cstate->copy_file == NULL)
1885 [ # # ]: 0 : ereport(ERROR,
1886 : : (errcode_for_file_access(),
1887 : : errmsg("could not execute command \"%s\": %m",
1888 : : cstate->filename)));
1889 : : }
1890 : : else
1891 : : {
1892 : : struct stat st;
1893 : :
1894 : 287 : progress_vals[1] = PROGRESS_COPY_TYPE_FILE;
1895 : 287 : cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_R);
1896 [ - + ]: 287 : if (cstate->copy_file == NULL)
1897 : : {
1898 : : /* copy errno because ereport subfunctions might change it */
1899 : 0 : int save_errno = errno;
1900 : :
1901 [ # # # # : 0 : ereport(ERROR,
# # ]
1902 : : (errcode_for_file_access(),
1903 : : errmsg("could not open file \"%s\" for reading: %m",
1904 : : cstate->filename),
1905 : : (save_errno == ENOENT || save_errno == EACCES) ?
1906 : : errhint("COPY FROM instructs the PostgreSQL server process to read a file. "
1907 : : "You may want a client-side facility such as psql's \\copy.") : 0));
1908 : : }
1909 : :
1910 [ - + ]: 287 : if (fstat(fileno(cstate->copy_file), &st))
1911 [ # # ]: 0 : ereport(ERROR,
1912 : : (errcode_for_file_access(),
1913 : : errmsg("could not stat file \"%s\": %m",
1914 : : cstate->filename)));
1915 : :
1916 [ - + ]: 287 : if (S_ISDIR(st.st_mode))
1917 [ # # ]: 0 : ereport(ERROR,
1918 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1919 : : errmsg("\"%s\" is a directory", cstate->filename)));
1920 : :
1921 : 287 : progress_vals[2] = st.st_size;
1922 : : }
1923 : : }
1924 : :
1925 : 1184 : pgstat_progress_update_multi_param(3, progress_cols, progress_vals);
1926 : :
1927 : 1184 : cstate->routine->CopyFromStart(cstate, tupDesc);
1928 : :
1929 : 1184 : MemoryContextSwitchTo(oldcontext);
1930 : :
1931 : 1184 : return cstate;
1932 : : }
1933 : :
1934 : : /*
1935 : : * Clean up storage and release resources for COPY FROM.
1936 : : */
1937 : : void
1938 : 991 : EndCopyFrom(CopyFromState cstate)
1939 : : {
1940 : : /* Invoke the end callback */
1941 : 991 : cstate->routine->CopyFromEnd(cstate);
1942 : :
1943 : : /* No COPY FROM related resources except memory. */
1944 [ - + ]: 991 : if (cstate->is_program)
1945 : : {
1946 : 0 : ClosePipeFromProgram(cstate);
1947 : : }
1948 : : else
1949 : : {
1950 [ + + - + ]: 991 : if (cstate->filename != NULL && FreeFile(cstate->copy_file))
1951 [ # # ]: 0 : ereport(ERROR,
1952 : : (errcode_for_file_access(),
1953 : : errmsg("could not close file \"%s\": %m",
1954 : : cstate->filename)));
1955 : : }
1956 : :
1957 : 991 : pgstat_progress_end_command();
1958 : :
1959 : 991 : MemoryContextDelete(cstate->copycontext);
1960 : 991 : pfree(cstate);
1961 : 991 : }
1962 : :
1963 : : /*
1964 : : * Closes the pipe from an external program, checking the pclose() return code.
1965 : : */
1966 : : static void
1967 : 0 : ClosePipeFromProgram(CopyFromState cstate)
1968 : : {
1969 : : int pclose_rc;
1970 : :
1971 : : Assert(cstate->is_program);
1972 : :
1973 : 0 : pclose_rc = ClosePipeStream(cstate->copy_file);
1974 [ # # ]: 0 : if (pclose_rc == -1)
1975 [ # # ]: 0 : ereport(ERROR,
1976 : : (errcode_for_file_access(),
1977 : : errmsg("could not close pipe to external command: %m")));
1978 [ # # ]: 0 : else if (pclose_rc != 0)
1979 : : {
1980 : : /*
1981 : : * If we ended a COPY FROM PROGRAM before reaching EOF, then it's
1982 : : * expectable for the called program to fail with SIGPIPE, and we
1983 : : * should not report that as an error. Otherwise, SIGPIPE indicates a
1984 : : * problem.
1985 : : */
1986 [ # # # # ]: 0 : if (!cstate->raw_reached_eof &&
1987 : 0 : wait_result_is_signal(pclose_rc, SIGPIPE))
1988 : 0 : return;
1989 : :
1990 [ # # ]: 0 : ereport(ERROR,
1991 : : (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
1992 : : errmsg("program \"%s\" failed",
1993 : : cstate->filename),
1994 : : errdetail_internal("%s", wait_result_to_str(pclose_rc))));
1995 : : }
1996 : : }
|
