LCOV - code coverage report
Current view: top level - src/backend/commands - copyfrom.c (source / functions) Hit Total Coverage
Test: PostgreSQL 17devel Lines: 505 562 89.9 %
Date: 2024-04-13 10:11:43 Functions: 16 17 94.1 %
Legend: Lines: hit not hit

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

Generated by: LCOV version 1.14