/*-------------------------------------------------------------------------
 *
 * copyfrom.c
 *  COPY <table> FROM file/program/client
 *
 * This file contains routines needed to efficiently load tuples into a
 * table.  That includes looking up the correct partition, firing triggers,
 * calling the table AM function to insert the data, and updating indexes.
 * Reading data from the input file or client and parsing it into Datums
 * is handled in copyfromparse.c.
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *   src/backend/commands/copyfrom.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <ctype.h>
#include <unistd.h>
#include <sys/stat.h>

#include "access/heapam.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/namespace.h"
#include "commands/copyapi.h"
#include "commands/copyfrom_internal.h"
#include "commands/progress.h"
#include "commands/trigger.h"
#include "executor/execPartition.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
#include "executor/tuptable.h"
#include "foreign/fdwapi.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/miscnodes.h"
#include "optimizer/optimizer.h"
#include "pgstat.h"
#include "rewrite/rewriteHandler.h"
#include "storage/fd.h"
#include "tcop/tcopprot.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/portal.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"

/*
 * No more than this many tuples per CopyMultiInsertBuffer
 *
 * Caution: Don't make this too big, as we could end up with this many
 * CopyMultiInsertBuffer items stored in CopyMultiInsertInfo's
 * multiInsertBuffers list.  Increasing this can cause quadratic growth in
 * memory requirements during copies into partitioned tables with a large
 * number of partitions.
 */
#define MAX_BUFFERED_TUPLES  1000

/*
 * Flush buffers if there are >= this many bytes, as counted by the input
 * size, of tuples stored.
 */
#define MAX_BUFFERED_BYTES  65535

/*
 * Trim the list of buffers back down to this number after flushing.  This
 * must be >= 2.
 */
#define MAX_PARTITION_BUFFERS 32

/* Stores multi-insert data related to a single relation in CopyFrom. */
typedef struct CopyMultiInsertBuffer
{
 TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */
 ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */
 BulkInsertState bistate; /* BulkInsertState for this rel if plain
 * table; NULL if foreign table */
 int   nused;   /* number of 'slots' containing tuples */
 uint64  linenos[MAX_BUFFERED_TUPLES]; /* Line # of tuple in copy
 * stream */
} CopyMultiInsertBuffer;

/*
 * Stores one or many CopyMultiInsertBuffers and details about the size and
 * number of tuples which are stored in them.  This allows multiple buffers to
 * exist at once when COPYing into a partitioned table.
 */
typedef struct CopyMultiInsertInfo
{
 List    *multiInsertBuffers; /* List of tracked CopyMultiInsertBuffers */
 int   bufferedTuples; /* number of tuples buffered over all buffers */
 int   bufferedBytes; /* number of bytes from all buffered tuples */
 CopyFromState cstate;  /* Copy state for this CopyMultiInsertInfo */
 EState    *estate;   /* Executor state used for COPY */
 CommandId mycid;   /* Command Id used for COPY */
 int   ti_options;  /* table insert options */
} CopyMultiInsertInfo;


/* non-export function prototypes */
static void ClosePipeFromProgram(CopyFromState cstate);

/*
 * Built-in format-specific routines. One-row callbacks are defined in
 * copyfromparse.c.
 */
static void CopyFromTextLikeInFunc(CopyFromState cstate, Oid atttypid, FmgrInfo *finfo,
           Oid *typioparam);
static void CopyFromTextLikeStart(CopyFromState cstate, TupleDesc tupDesc);
static void CopyFromTextLikeEnd(CopyFromState cstate);
static void CopyFromBinaryInFunc(CopyFromState cstate, Oid atttypid,
         FmgrInfo *finfo, Oid *typioparam);
static void CopyFromBinaryStart(CopyFromState cstate, TupleDesc tupDesc);
static void CopyFromBinaryEnd(CopyFromState cstate);


/*
 * COPY FROM routines for built-in formats.
 *
 * CSV and text formats share the same TextLike routines except for the
 * one-row callback.
 */

/* text format */
static const CopyFromRoutine CopyFromRoutineText = {
 .CopyFromInFunc = CopyFromTextLikeInFunc,
 .CopyFromStart = CopyFromTextLikeStart,
 .CopyFromOneRow = CopyFromTextOneRow,
 .CopyFromEnd = CopyFromTextLikeEnd,
};

/* CSV format */
static const CopyFromRoutine CopyFromRoutineCSV = {
 .CopyFromInFunc = CopyFromTextLikeInFunc,
 .CopyFromStart = CopyFromTextLikeStart,
 .CopyFromOneRow = CopyFromCSVOneRow,
 .CopyFromEnd = CopyFromTextLikeEnd,
};

/* binary format */
static const CopyFromRoutine CopyFromRoutineBinary = {
 .CopyFromInFunc = CopyFromBinaryInFunc,
 .CopyFromStart = CopyFromBinaryStart,
 .CopyFromOneRow = CopyFromBinaryOneRow,
 .CopyFromEnd = CopyFromBinaryEnd,
};

/* Return a COPY FROM routine for the given options */
static const CopyFromRoutine *
CopyFromGetRoutine(const CopyFormatOptions *opts)
{
 if (opts->csv_mode)
  return &CopyFromRoutineCSV;
 else if (opts->binary)
  return &CopyFromRoutineBinary;

 /* default is text */
 return &CopyFromRoutineText;
}

/* Implementation of the start callback for text and CSV formats */
static void
CopyFromTextLikeStart(CopyFromState cstate, TupleDesc tupDesc)
{
 AttrNumber attr_count;

 /*
  * If encoding conversion is needed, we need another buffer to hold the
  * converted input data.  Otherwise, we can just point input_buf to the
  * same buffer as raw_buf.
 */
 if (cstate->need_transcoding)
 {
  cstate->input_buf = (char *) palloc(INPUT_BUF_SIZE + 1);
  cstate->input_buf_index = cstate->input_buf_len = 0;
 }
 else
  cstate->input_buf = cstate->raw_buf;
 cstate->input_reached_eof = false;

 initStringInfo(&cstate->line_buf);

 /*
  * Create workspace for CopyReadAttributes results; used by CSV and text
  * format.
 */
 attr_count = list_length(cstate->attnumlist);
 cstate->max_fields = attr_count;
 cstate->raw_fields = (char **) palloc(attr_count * sizeof(char *));
}

/*
 * Implementation of the infunc callback for text and CSV formats. Assign
 * the input function data to the given *finfo.
 */
static void
CopyFromTextLikeInFunc(CopyFromState cstate, Oid atttypid, FmgrInfo *finfo,
        Oid *typioparam)
{
 Oid   func_oid;

 getTypeInputInfo(atttypid, &func_oid, typioparam);
 fmgr_info(func_oid, finfo);
}

/* Implementation of the end callback for text and CSV formats */
static void
CopyFromTextLikeEnd(CopyFromState cstate)
{
 /* nothing to do */
}

/* Implementation of the start callback for binary format */
static void
CopyFromBinaryStart(CopyFromState cstate, TupleDesc tupDesc)
{
 /* Read and verify binary header */
 ReceiveCopyBinaryHeader(cstate);
}

/*
 * Implementation of the infunc callback for binary format. Assign
 * the binary input function to the given *finfo.
 */
static void
CopyFromBinaryInFunc(CopyFromState cstate, Oid atttypid,
      FmgrInfo *finfo, Oid *typioparam)
{
 Oid   func_oid;

 getTypeBinaryInputInfo(atttypid, &func_oid, typioparam);
 fmgr_info(func_oid, finfo);
}

/* Implementation of the end callback for binary format */
static void
CopyFromBinaryEnd(CopyFromState cstate)
{
 /* nothing to do */
}

/*
 * error context callback for COPY FROM
 *
 * The argument for the error context must be CopyFromState.
 */
void
CopyFromErrorCallback(void *arg)
{
 CopyFromState cstate = (CopyFromState) arg;

 if (cstate->relname_only)
 {
  errcontext("COPY %s",
       cstate->cur_relname);
  return;
 }
 if (cstate->opts.binary)
 {
  /* can't usefully display the data */
  if (cstate->cur_attname)
   errcontext("COPY %s, line %" PRIu64 ", column %s",
        cstate->cur_relname,
        cstate->cur_lineno,
        cstate->cur_attname);
  else
   errcontext("COPY %s, line %" PRIu64,
        cstate->cur_relname,
        cstate->cur_lineno);
 }
 else
 {
  if (cstate->cur_attname && cstate->cur_attval)
  {
   /* error is relevant to a particular column */
   char    *attval;

   attval = CopyLimitPrintoutLength(cstate->cur_attval);
   errcontext("COPY %s, line %" PRIu64 ", column %s: \"%s\"",
        cstate->cur_relname,
        cstate->cur_lineno,
        cstate->cur_attname,
        attval);
   pfree(attval);
  }
  else if (cstate->cur_attname)
  {
   /* error is relevant to a particular column, value is NULL */
   errcontext("COPY %s, line %" PRIu64 ", column %s: null input",
        cstate->cur_relname,
        cstate->cur_lineno,
        cstate->cur_attname);
  }
  else
  {
   /*
    * Error is relevant to a particular line.
    *
    * If line_buf still contains the correct line, print it.
 */
   if (cstate->line_buf_valid)
   {
    char    *lineval;

    lineval = CopyLimitPrintoutLength(cstate->line_buf.data);
    errcontext("COPY %s, line %" PRIu64 ": \"%s\"",
         cstate->cur_relname,
         cstate->cur_lineno, lineval);
    pfree(lineval);
   }
   else
   {
    errcontext("COPY %s, line %" PRIu64,
         cstate->cur_relname,
         cstate->cur_lineno);
   }
  }
 }
}

/*
 * Make sure we don't print an unreasonable amount of COPY data in a message.
 *
 * Returns a pstrdup'd copy of the input.
 */
char *
CopyLimitPrintoutLength(const char *str)
{
#define MAX_COPY_DATA_DISPLAY 100

 int   slen = strlen(str);
 int   len;
 char    *res;

 /* Fast path if definitely okay */
 if (slen <= MAX_COPY_DATA_DISPLAY)
  return pstrdup(str);

 /* Apply encoding-dependent truncation */
 len = pg_mbcliplen(str, slen, MAX_COPY_DATA_DISPLAY);

 /*
  * Truncate, and add "..." to show we truncated the input.
 */
 res = (char *) palloc(len + 4);
 memcpy(res, str, len);
 strcpy(res + len, "...");

 return res;
}

/*
 * Allocate memory and initialize a new CopyMultiInsertBuffer for this
 * ResultRelInfo.
 */
static CopyMultiInsertBuffer *
CopyMultiInsertBufferInit(ResultRelInfo *rri)
{
 CopyMultiInsertBuffer *buffer;

 buffer = (CopyMultiInsertBuffer *) palloc(sizeof(CopyMultiInsertBuffer));
 memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES);
 buffer->resultRelInfo = rri;
 buffer->bistate = (rri->ri_FdwRoutine == NULL) ? GetBulkInsertState() : NULL;
 buffer->nused = 0;

 return buffer;
}

/*
 * Make a new buffer for this ResultRelInfo.
 */
static inline void
CopyMultiInsertInfoSetupBuffer(CopyMultiInsertInfo *miinfo,
          ResultRelInfo *rri)
{
 CopyMultiInsertBuffer *buffer;

 buffer = CopyMultiInsertBufferInit(rri);

 /* Setup back-link so we can easily find this buffer again */
 rri->ri_CopyMultiInsertBuffer = buffer;
 /* Record that we're tracking this buffer */
 miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
}

/*
 * Initialize an already allocated CopyMultiInsertInfo.
 *
 * If rri is a non-partitioned table then a CopyMultiInsertBuffer is set up
 * for that table.
 */
static void
CopyMultiInsertInfoInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
      CopyFromState cstate, EState *estate, CommandId mycid,
      int ti_options)
{
 miinfo->multiInsertBuffers = NIL;
 miinfo->bufferedTuples = 0;
 miinfo->bufferedBytes = 0;
 miinfo->cstate = cstate;
 miinfo->estate = estate;
 miinfo->mycid = mycid;
 miinfo->ti_options = ti_options;

 /*
  * Only setup the buffer when not dealing with a partitioned table.
  * Buffers for partitioned tables will just be setup when we need to send
  * tuples their way for the first time.
 */
 if (rri->ri_RelationDesc->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
  CopyMultiInsertInfoSetupBuffer(miinfo, rri);
}

/*
 * Returns true if the buffers are full
 */
static inline bool
CopyMultiInsertInfoIsFull(CopyMultiInsertInfo *miinfo)
{
 if (miinfo->bufferedTuples >= MAX_BUFFERED_TUPLES ||
  miinfo->bufferedBytes >= MAX_BUFFERED_BYTES)
  return true;
 return false;
}

/*
 * Returns true if we have no buffered tuples
 */
static inline bool
CopyMultiInsertInfoIsEmpty(CopyMultiInsertInfo *miinfo)
{
 return miinfo->bufferedTuples == 0;
}

/*
 * Write the tuples stored in 'buffer' out to the table.
 */
static inline void
CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo,
         CopyMultiInsertBuffer *buffer,
         int64 *processed)
{
 CopyFromState cstate = miinfo->cstate;
 EState    *estate = miinfo->estate;
 int   nused = buffer->nused;
 ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
 TupleTableSlot **slots = buffer->slots;
 int   i;

 if (resultRelInfo->ri_FdwRoutine)
 {
  int   batch_size = resultRelInfo->ri_BatchSize;
  int   sent = 0;

  Assert(buffer->bistate == NULL);

  /* Ensure that the FDW supports batching and it's enabled */
  Assert(resultRelInfo->ri_FdwRoutine->ExecForeignBatchInsert);
  Assert(batch_size > 1);

  /*
   * We suppress error context information other than the relation name,
   * if one of the operations below fails.
 */
  Assert(!cstate->relname_only);
  cstate->relname_only = true;

  while (sent < nused)
  {
   int   size = (batch_size < nused - sent) ? batch_size : (nused - sent);
   int   inserted = size;
   TupleTableSlot **rslots;

   /* insert into foreign table: let the FDW do it */
   rslots =
    resultRelInfo->ri_FdwRoutine->ExecForeignBatchInsert(estate,
                  resultRelInfo,
                  &slots[sent],
                  NULL,
                  &inserted);

   sent += size;

   /* No need to do anything if there are no inserted rows */
   if (inserted <= 0)
    continue;

   /* Triggers on foreign tables should not have transition tables */
   Assert(resultRelInfo->ri_TrigDesc == NULL ||
       resultRelInfo->ri_TrigDesc->trig_insert_new_table == false);

   /* Run AFTER ROW INSERT triggers */
   if (resultRelInfo->ri_TrigDesc != NULL &&
    resultRelInfo->ri_TrigDesc->trig_insert_after_row)
   {
    Oid   relid = RelationGetRelid(resultRelInfo->ri_RelationDesc);

    for (i = 0; i < inserted; i++)
    {
     TupleTableSlot *slot = rslots[i];

     /*
      * AFTER ROW Triggers might reference the tableoid column,
      * so (re-)initialize tts_tableOid before evaluating them.
 */
     slot->tts_tableOid = relid;

     ExecARInsertTriggers(estate, resultRelInfo,
           slot, NIL,
           cstate->transition_capture);
    }
   }

   /* Update the row counter and progress of the COPY command */
   *processed += inserted;
   pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
           *processed);
  }

  for (i = 0; i < nused; i++)
   ExecClearTuple(slots[i]);

  /* reset relname_only */
  cstate->relname_only = false;
 }
 else
 {
  CommandId mycid = miinfo->mycid;
  int   ti_options = miinfo->ti_options;
  bool  line_buf_valid = cstate->line_buf_valid;
  uint64  save_cur_lineno = cstate->cur_lineno;
  MemoryContext oldcontext;

  Assert(buffer->bistate != NULL);

  /*
   * Print error context information correctly, if one of the operations
   * below fails.
 */
  cstate->line_buf_valid = false;

  /*
   * table_multi_insert may leak memory, so switch to short-lived memory
   * context before calling it.
 */
  oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
  table_multi_insert(resultRelInfo->ri_RelationDesc,
         slots,
         nused,
         mycid,
         ti_options,
         buffer->bistate);
  MemoryContextSwitchTo(oldcontext);

  for (i = 0; i < nused; i++)
  {
   /*
    * If there are any indexes, update them for all the inserted
    * tuples, and run AFTER ROW INSERT triggers.
 */
   if (resultRelInfo->ri_NumIndices > 0)
   {
    List    *recheckIndexes;

    cstate->cur_lineno = buffer->linenos[i];
    recheckIndexes =
     ExecInsertIndexTuples(resultRelInfo,
            buffer->slots[i], estate, false,
            false, NULL, NIL, false);
    ExecARInsertTriggers(estate, resultRelInfo,
          slots[i], recheckIndexes,
          cstate->transition_capture);
    list_free(recheckIndexes);
   }

   /*
    * There's no indexes, but see if we need to run AFTER ROW INSERT
    * triggers anyway.
 */
   else if (resultRelInfo->ri_TrigDesc != NULL &&
      (resultRelInfo->ri_TrigDesc->trig_insert_after_row ||
       resultRelInfo->ri_TrigDesc->trig_insert_new_table))
   {
    cstate->cur_lineno = buffer->linenos[i];
    ExecARInsertTriggers(estate, resultRelInfo,
          slots[i], NIL,
          cstate->transition_capture);
   }

   ExecClearTuple(slots[i]);
  }

  /* Update the row counter and progress of the COPY command */
  *processed += nused;
  pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
          *processed);

  /* reset cur_lineno and line_buf_valid to what they were */
  cstate->line_buf_valid = line_buf_valid;
  cstate->cur_lineno = save_cur_lineno;
 }

 /* Mark that all slots are free */
 buffer->nused = 0;
}

/*
 * Drop used slots and free member for this buffer.
 *
 * The buffer must be flushed before cleanup.
 */
static inline void
CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
        CopyMultiInsertBuffer *buffer)
{
 ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
 int   i;

 /* Ensure buffer was flushed */
 Assert(buffer->nused == 0);

 /* Remove back-link to ourself */
 resultRelInfo->ri_CopyMultiInsertBuffer = NULL;

 if (resultRelInfo->ri_FdwRoutine == NULL)
 {
  Assert(buffer->bistate != NULL);
  FreeBulkInsertState(buffer->bistate);
 }
 else
  Assert(buffer->bistate == NULL);

 /* Since we only create slots on demand, just drop the non-null ones. */
 for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++)
  ExecDropSingleTupleTableSlot(buffer->slots[i]);

 if (resultRelInfo->ri_FdwRoutine == NULL)
  table_finish_bulk_insert(resultRelInfo->ri_RelationDesc,
         miinfo->ti_options);

 pfree(buffer);
}

/*
 * Write out all stored tuples in all buffers out to the tables.
 *
 * Once flushed we also trim the tracked buffers list down to size by removing
 * the buffers created earliest first.
 *
 * Callers should pass 'curr_rri' as the ResultRelInfo that's currently being
 * used.  When cleaning up old buffers we'll never remove the one for
 * 'curr_rri'.
 */
static inline void
CopyMultiInsertInfoFlush(CopyMultiInsertInfo *miinfo, ResultRelInfo *curr_rri,
       int64 *processed)
{
 ListCell   *lc;

 foreach(lc, miinfo->multiInsertBuffers)
 {
  CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) lfirst(lc);

  CopyMultiInsertBufferFlush(miinfo, buffer, processed);
 }

 miinfo->bufferedTuples = 0;
 miinfo->bufferedBytes = 0;

 /*
  * Trim the list of tracked buffers down if it exceeds the limit.  Here we
  * remove buffers starting with the ones we created first.  It seems less
  * likely that these older ones will be needed than the ones that were
  * just created.
 */
 while (list_length(miinfo->multiInsertBuffers) > MAX_PARTITION_BUFFERS)
 {
  CopyMultiInsertBuffer *buffer;

  buffer = (CopyMultiInsertBuffer *) linitial(miinfo->multiInsertBuffers);

  /*
   * We never want to remove the buffer that's currently being used, so
   * if we happen to find that then move it to the end of the list.
 */
  if (buffer->resultRelInfo == curr_rri)
  {
   /*
    * The code below would misbehave if we were trying to reduce the
    * list to less than two items.
 */
   StaticAssertDecl(MAX_PARTITION_BUFFERS >= 2,
        "MAX_PARTITION_BUFFERS must be >= 2");

   miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
   miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
   buffer = (CopyMultiInsertBuffer *) linitial(miinfo->multiInsertBuffers);
  }

  CopyMultiInsertBufferCleanup(miinfo, buffer);
  miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
 }
}

/*
 * Cleanup allocated buffers and free memory
 */
static inline void
CopyMultiInsertInfoCleanup(CopyMultiInsertInfo *miinfo)
{
 ListCell   *lc;

 foreach(lc, miinfo->multiInsertBuffers)
  CopyMultiInsertBufferCleanup(miinfo, lfirst(lc));

 list_free(miinfo->multiInsertBuffers);
}

/*
 * Get the next TupleTableSlot that the next tuple should be stored in.
 *
 * Callers must ensure that the buffer is not full.
 *
 * Note: 'miinfo' is unused but has been included for consistency with the
 * other functions in this area.
 */
static inline TupleTableSlot *
CopyMultiInsertInfoNextFreeSlot(CopyMultiInsertInfo *miinfo,
        ResultRelInfo *rri)
{
 CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer;
 int   nused;

 Assert(buffer != NULL);
 Assert(buffer->nused < MAX_BUFFERED_TUPLES);

 nused = buffer->nused;

 if (buffer->slots[nused] == NULL)
  buffer->slots[nused] = table_slot_create(rri->ri_RelationDesc, NULL);
 return buffer->slots[nused];
}

/*
 * Record the previously reserved TupleTableSlot that was reserved by
 * CopyMultiInsertInfoNextFreeSlot as being consumed.
 */
static inline void
CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
       TupleTableSlot *slot, int tuplen, uint64 lineno)
{
 CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer;

 Assert(buffer != NULL);
 Assert(slot == buffer->slots[buffer->nused]);

 /* Store the line number so we can properly report any errors later */
 buffer->linenos[buffer->nused] = lineno;

 /* Record this slot as being used */
 buffer->nused++;

 /* Update how many tuples are stored and their size */
 miinfo->bufferedTuples++;
 miinfo->bufferedBytes += tuplen;
}

/*
 * Copy FROM file to relation.
 */
uint64
CopyFrom(CopyFromState cstate)
{
 ResultRelInfo *resultRelInfo;
 ResultRelInfo *target_resultRelInfo;
 ResultRelInfo *prevResultRelInfo = NULL;
 EState    *estate = CreateExecutorState(); /* for ExecConstraints() */
 ModifyTableState *mtstate;
 ExprContext *econtext;
 TupleTableSlot *singleslot = NULL;
 MemoryContext oldcontext = CurrentMemoryContext;

 PartitionTupleRouting *proute = NULL;
 ErrorContextCallback errcallback;
 CommandId mycid = GetCurrentCommandId(true);
 int   ti_options = 0; /* start with default options for insert */
 BulkInsertState bistate = NULL;
 CopyInsertMethod insertMethod;
 CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
 int64  processed = 0;
 int64  excluded = 0;
 bool  has_before_insert_row_trig;
 bool  has_instead_insert_row_trig;
 bool  leafpart_use_multi_insert = false;

 Assert(cstate->rel);
 Assert(list_length(cstate->range_table) == 1);

 if (cstate->opts.on_error != COPY_ON_ERROR_STOP)
  Assert(cstate->escontext);

 /*
  * The target must be a plain, foreign, or partitioned relation, or have
  * an INSTEAD OF INSERT row trigger.  (Currently, such triggers are only
  * allowed on views, so we only hint about them in the view case.)
 */
 if (cstate->rel->rd_rel->relkind != RELKIND_RELATION &&
  cstate->rel->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
  cstate->rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE &&
  !(cstate->rel->trigdesc &&
    cstate->rel->trigdesc->trig_insert_instead_row))
 {
  if (cstate->rel->rd_rel->relkind == RELKIND_VIEW)
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("cannot copy to view \"%s\"",
       RelationGetRelationName(cstate->rel)),
      errhint("To enable copying to a view, provide an INSTEAD OF INSERT trigger.")));
  else if (cstate->rel->rd_rel->relkind == RELKIND_MATVIEW)
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("cannot copy to materialized view \"%s\"",
       RelationGetRelationName(cstate->rel))));
  else if (cstate->rel->rd_rel->relkind == RELKIND_SEQUENCE)
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("cannot copy to sequence \"%s\"",
       RelationGetRelationName(cstate->rel))));
  else
   ereport(ERROR,
     (errcode(ERRCODE_WRONG_OBJECT_TYPE),
      errmsg("cannot copy to non-table relation \"%s\"",
       RelationGetRelationName(cstate->rel))));
 }

 /*
  * If the target file is new-in-transaction, we assume that checking FSM
  * for free space is a waste of time.  This could possibly be wrong, but
  * it's unlikely.
 */
 if (RELKIND_HAS_STORAGE(cstate->rel->rd_rel->relkind) &&
  (cstate->rel->rd_createSubid != InvalidSubTransactionId ||
   cstate->rel->rd_firstRelfilelocatorSubid != InvalidSubTransactionId))
  ti_options |= TABLE_INSERT_SKIP_FSM;

 /*
  * Optimize if new relation storage was created in this subxact or one of
  * its committed children and we won't see those rows later as part of an
  * earlier scan or command. The subxact test ensures that if this subxact
  * aborts then the frozen rows won't be visible after xact cleanup.  Note
  * that the stronger test of exactly which subtransaction created it is
  * crucial for correctness of this optimization. The test for an earlier
  * scan or command tolerates false negatives. FREEZE causes other sessions
  * to see rows they would not see under MVCC, and a false negative merely
  * spreads that anomaly to the current session.
 */
 if (cstate->opts.freeze)
 {
  /*
   * We currently disallow COPY FREEZE on partitioned tables.  The
   * reason for this is that we've simply not yet opened the partitions
   * to determine if the optimization can be applied to them.  We could
   * go and open them all here, but doing so may be quite a costly
   * overhead for small copies.  In any case, we may just end up routing
   * tuples to a small number of partitions.  It seems better just to
   * raise an ERROR for partitioned tables.
 */
  if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
  {
   ereport(ERROR,
     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
      errmsg("cannot perform COPY FREEZE on a partitioned table")));
  }

  /* There's currently no support for COPY FREEZE on foreign tables. */
  if (cstate->rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
   ereport(ERROR,
     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
      errmsg("cannot perform COPY FREEZE on a foreign table")));

  /*
   * Tolerate one registration for the benefit of FirstXactSnapshot.
   * Scan-bearing queries generally create at least two registrations,
   * though relying on that is fragile, as is ignoring ActiveSnapshot.
   * Clear CatalogSnapshot to avoid counting its registration.  We'll
   * still detect ongoing catalog scans, each of which separately
   * registers the snapshot it uses.
 */
  InvalidateCatalogSnapshot();
  if (!ThereAreNoPriorRegisteredSnapshots() || !ThereAreNoReadyPortals())
   ereport(ERROR,
     (errcode(ERRCODE_INVALID_TRANSACTION_STATE),
      errmsg("cannot perform COPY FREEZE because of prior transaction activity")));

  if (cstate->rel->rd_createSubid != GetCurrentSubTransactionId() &&
   cstate->rel->rd_newRelfilelocatorSubid != GetCurrentSubTransactionId())
   ereport(ERROR,
     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      errmsg("cannot perform COPY FREEZE because the table was not created or truncated in the current subtransaction")));

  ti_options |= TABLE_INSERT_FROZEN;
 }

 /*
  * We need a ResultRelInfo so we can use the regular executor's
  * index-entry-making machinery.  (There used to be a huge amount of code
  * here that basically duplicated execUtils.c ...)
 */
 ExecInitRangeTable(estate, cstate->range_table, cstate->rteperminfos,
        bms_make_singleton(1));
 resultRelInfo = target_resultRelInfo = makeNode(ResultRelInfo);
 ExecInitResultRelation(estate, resultRelInfo, 1);

 /* Verify the named relation is a valid target for INSERT */
 CheckValidResultRel(resultRelInfo, CMD_INSERT, ONCONFLICT_NONE, NIL);

 ExecOpenIndices(resultRelInfo, false);

 /*
  * Set up a ModifyTableState so we can let FDW(s) init themselves for
  * foreign-table result relation(s).
 */
 mtstate = makeNode(ModifyTableState);
 mtstate->ps.plan = NULL;
 mtstate->ps.state = estate;
 mtstate->operation = CMD_INSERT;
 mtstate->mt_nrels = 1;
 mtstate->resultRelInfo = resultRelInfo;
 mtstate->rootResultRelInfo = resultRelInfo;

 if (resultRelInfo->ri_FdwRoutine != NULL &&
  resultRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL)
  resultRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate,
               resultRelInfo);

 /*
  * Also, if the named relation is a foreign table, determine if the FDW
  * supports batch insert and determine the batch size (a FDW may support
  * batching, but it may be disabled for the server/table).
  *
  * If the FDW does not support batching, we set the batch size to 1.
 */
 if (resultRelInfo->ri_FdwRoutine != NULL &&
  resultRelInfo->ri_FdwRoutine->GetForeignModifyBatchSize &&
  resultRelInfo->ri_FdwRoutine->ExecForeignBatchInsert)
  resultRelInfo->ri_BatchSize =
   resultRelInfo->ri_FdwRoutine->GetForeignModifyBatchSize(resultRelInfo);
 else
  resultRelInfo->ri_BatchSize = 1;

 Assert(resultRelInfo->ri_BatchSize >= 1);

 /* Prepare to catch AFTER triggers. */
 AfterTriggerBeginQuery();

 /*
  * If there are any triggers with transition tables on the named relation,
  * we need to be prepared to capture transition tuples.
  *
  * Because partition tuple routing would like to know about whether
  * transition capture is active, we also set it in mtstate, which is
  * passed to ExecFindPartition() below.
 */
 cstate->transition_capture = mtstate->mt_transition_capture =
  MakeTransitionCaptureState(cstate->rel->trigdesc,
           RelationGetRelid(cstate->rel),
           CMD_INSERT);

 /*
  * If the named relation is a partitioned table, initialize state for
  * CopyFrom tuple routing.
 */
 if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
  proute = ExecSetupPartitionTupleRouting(estate, cstate->rel);

 if (cstate->whereClause)
  cstate->qualexpr = ExecInitQual(castNode(List, cstate->whereClause),
          &mtstate->ps);

 /*
  * It's generally more efficient to prepare a bunch of tuples for
  * insertion, and insert them in one
  * table_multi_insert()/ExecForeignBatchInsert() call, than call
  * table_tuple_insert()/ExecForeignInsert() separately for every tuple.
  * However, there are a number of reasons why we might not be able to do
  * this.  These are explained below.
 */
 if (resultRelInfo->ri_TrigDesc != NULL &&
  (resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
   resultRelInfo->ri_TrigDesc->trig_insert_instead_row))
 {
  /*
   * Can't support multi-inserts when there are any BEFORE/INSTEAD OF
   * triggers on the table. Such triggers might query the table we're
   * inserting into and act differently if the tuples that have already
   * been processed and prepared for insertion are not there.
 */
  insertMethod = CIM_SINGLE;
 }
 else if (resultRelInfo->ri_FdwRoutine != NULL &&
    resultRelInfo->ri_BatchSize == 1)
 {
  /*
   * Can't support multi-inserts to a foreign table if the FDW does not
   * support batching, or it's disabled for the server or foreign table.
 */
  insertMethod = CIM_SINGLE;
 }
 else if (proute != NULL && resultRelInfo->ri_TrigDesc != NULL &&
    resultRelInfo->ri_TrigDesc->trig_insert_new_table)
 {
  /*
   * For partitioned tables we can't support multi-inserts when there
   * are any statement level insert triggers. It might be possible to
   * allow partitioned tables with such triggers in the future, but for
   * now, CopyMultiInsertInfoFlush expects that any after row insert and
   * statement level insert triggers are on the same relation.
 */
  insertMethod = CIM_SINGLE;
 }
 else if (cstate->volatile_defexprs)
 {
  /*
   * Can't support multi-inserts if there are any volatile default
   * expressions in the table.  Similarly to the trigger case above,
   * such expressions may query the table we're inserting into.
   *
   * Note: It does not matter if any partitions have any volatile
   * default expressions as we use the defaults from the target of the
   * COPY command.
 */
  insertMethod = CIM_SINGLE;
 }
 else if (contain_volatile_functions(cstate->whereClause))
 {
  /*
   * Can't support multi-inserts if there are any volatile function
   * expressions in WHERE clause.  Similarly to the trigger case above,
   * such expressions may query the table we're inserting into.
   *
   * Note: the whereClause was already preprocessed in DoCopy(), so it's
   * okay to use contain_volatile_functions() directly.
 */
  insertMethod = CIM_SINGLE;
 }
 else
 {
  /*
   * For partitioned tables, we may still be able to perform bulk
   * inserts.  However, the possibility of this depends on which types
   * of triggers exist on the partition.  We must disable bulk inserts
   * if the partition is a foreign table that can't use batching or it
   * has any before row insert or insert instead triggers (same as we
   * checked above for the parent table).  Since the partition's
   * resultRelInfos are initialized only when we actually need to insert
   * the first tuple into them, we must have the intermediate insert
   * method of CIM_MULTI_CONDITIONAL to flag that we must later
   * determine if we can use bulk-inserts for the partition being
   * inserted into.
 */
  if (proute)
   insertMethod = CIM_MULTI_CONDITIONAL;
  else
   insertMethod = CIM_MULTI;

  CopyMultiInsertInfoInit(&multiInsertInfo, resultRelInfo, cstate,
        estate, mycid, ti_options);
 }

 /*
  * If not using batch mode (which allocates slots as needed) set up a
  * tuple slot too. When inserting into a partitioned table, we also need
  * one, even if we might batch insert, to read the tuple in the root
  * partition's form.
 */
 if (insertMethod == CIM_SINGLE || insertMethod == CIM_MULTI_CONDITIONAL)
 {
  singleslot = table_slot_create(resultRelInfo->ri_RelationDesc,
            &estate->es_tupleTable);
  bistate = GetBulkInsertState();
 }

 has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
          resultRelInfo->ri_TrigDesc->trig_insert_before_row);

 has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
           resultRelInfo->ri_TrigDesc->trig_insert_instead_row);

 /*
  * Check BEFORE STATEMENT insertion triggers. It's debatable whether we
  * should do this for COPY, since it's not really an "INSERT" statement as
  * such. However, executing these triggers maintains consistency with the
  * EACH ROW triggers that we already fire on COPY.
 */
 ExecBSInsertTriggers(estate, resultRelInfo);

 econtext = GetPerTupleExprContext(estate);

 /* Set up callback to identify error line number */
 errcallback.callback = CopyFromErrorCallback;
 errcallback.arg = cstate;
 errcallback.previous = error_context_stack;
 error_context_stack = &errcallback;

 for (;;)
 {
  TupleTableSlot *myslot;
  bool  skip_tuple;

  CHECK_FOR_INTERRUPTS();

  /*
   * Reset the per-tuple exprcontext. We do this after every tuple, to
   * clean-up after expression evaluations etc.
 */
  ResetPerTupleExprContext(estate);

  /* select slot to (initially) load row into */
  if (insertMethod == CIM_SINGLE || proute)
  {
   myslot = singleslot;
   Assert(myslot != NULL);
  }
  else
  {
   Assert(resultRelInfo == target_resultRelInfo);
   Assert(insertMethod == CIM_MULTI);

   myslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
              resultRelInfo);
  }

  /*
   * Switch to per-tuple context before calling NextCopyFrom, which does
   * evaluate default expressions etc. and requires per-tuple context.
 */
  MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));

  ExecClearTuple(myslot);

  /* Directly store the values/nulls array in the slot */
  if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
   break;

  if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE &&
   cstate->escontext->error_occurred)
  {
   /*
    * Soft error occurred, skip this tuple and just make
    * ErrorSaveContext ready for the next NextCopyFrom. Since we
    * don't set details_wanted and error_data is not to be filled,
    * just resetting error_occurred is enough.
 */
   cstate->escontext->error_occurred = false;

   /* Report that this tuple was skipped by the ON_ERROR clause */
   pgstat_progress_update_param(PROGRESS_COPY_TUPLES_SKIPPED,
           cstate->num_errors);

   if (cstate->opts.reject_limit > 0 &&
    cstate->num_errors > cstate->opts.reject_limit)
    ereport(ERROR,
      (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
       errmsg("skipped more than REJECT_LIMIT (%" PRId64 ") rows due to data type incompatibility",
        cstate->opts.reject_limit)));

   /* Repeat NextCopyFrom() until no soft error occurs */
   continue;
  }

  ExecStoreVirtualTuple(myslot);

  /*
   * Constraints and where clause might reference the tableoid column,
   * so (re-)initialize tts_tableOid before evaluating them.
 */
  myslot->tts_tableOid = RelationGetRelid(target_resultRelInfo->ri_RelationDesc);

  /* Triggers and stuff need to be invoked in query context. */
  MemoryContextSwitchTo(oldcontext);

  if (cstate->whereClause)
  {
   econtext->ecxt_scantuple = myslot;
   /* Skip items that don't match COPY's WHERE clause */
   if (!ExecQual(cstate->qualexpr, econtext))
   {
    /*
     * Report that this tuple was filtered out by the WHERE
     * clause.
 */
    pgstat_progress_update_param(PROGRESS_COPY_TUPLES_EXCLUDED,
            ++excluded);
    continue;
   }
  }

  /* Determine the partition to insert the tuple into */
  if (proute)
  {
   TupleConversionMap *map;

   /*
    * Attempt to find a partition suitable for this tuple.
    * ExecFindPartition() will raise an error if none can be found or
    * if the found partition is not suitable for INSERTs.
 */
   resultRelInfo = ExecFindPartition(mtstate, target_resultRelInfo,
             proute, myslot, estate);

   if (prevResultRelInfo != resultRelInfo)
   {
    /* Determine which triggers exist on this partition */
    has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
             resultRelInfo->ri_TrigDesc->trig_insert_before_row);

    has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
              resultRelInfo->ri_TrigDesc->trig_insert_instead_row);

    /*
     * Disable multi-inserts when the partition has BEFORE/INSTEAD
     * OF triggers, or if the partition is a foreign table that
     * can't use batching.
 */
    leafpart_use_multi_insert = insertMethod == CIM_MULTI_CONDITIONAL &&
     !has_before_insert_row_trig &&
     !has_instead_insert_row_trig &&
     (resultRelInfo->ri_FdwRoutine == NULL ||
      resultRelInfo->ri_BatchSize > 1);

    /* Set the multi-insert buffer to use for this partition. */
    if (leafpart_use_multi_insert)
    {
     if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL)
      CopyMultiInsertInfoSetupBuffer(&multiInsertInfo,
                resultRelInfo);
    }
    else if (insertMethod == CIM_MULTI_CONDITIONAL &&
       !CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
    {
     /*
      * Flush pending inserts if this partition can't use
      * batching, so rows are visible to triggers etc.
 */
     CopyMultiInsertInfoFlush(&multiInsertInfo,
            resultRelInfo,
            &processed);
    }

    if (bistate != NULL)
     ReleaseBulkInsertStatePin(bistate);
    prevResultRelInfo = resultRelInfo;
   }

   /*
    * If we're capturing transition tuples, we might need to convert
    * from the partition rowtype to root rowtype. But if there are no
    * BEFORE triggers on the partition that could change the tuple,
    * we can just remember the original unconverted tuple to avoid a
    * needless round trip conversion.
 */
   if (cstate->transition_capture != NULL)
    cstate->transition_capture->tcs_original_insert_tuple =
     !has_before_insert_row_trig ? myslot : NULL;

   /*
    * We might need to convert from the root rowtype to the partition
    * rowtype.
 */
   map = ExecGetRootToChildMap(resultRelInfo, estate);
   if (insertMethod == CIM_SINGLE || !leafpart_use_multi_insert)
   {
    /* non batch insert */
    if (map != NULL)
    {
     TupleTableSlot *new_slot;

     new_slot = resultRelInfo->ri_PartitionTupleSlot;
     myslot = execute_attr_map_slot(map->attrMap, myslot, new_slot);
    }
   }
   else
   {
    /*
     * Prepare to queue up tuple for later batch insert into
     * current partition.
 */
    TupleTableSlot *batchslot;

    /* no other path available for partitioned table */
    Assert(insertMethod == CIM_MULTI_CONDITIONAL);

    batchslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
               resultRelInfo);

    if (map != NULL)
     myslot = execute_attr_map_slot(map->attrMap, myslot,
               batchslot);
    else
    {
     /*
      * This looks more expensive than it is (Believe me, I
      * optimized it away. Twice.). The input is in virtual
      * form, and we'll materialize the slot below - for most
      * slot types the copy performs the work materialization
      * would later require anyway.
 */
     ExecCopySlot(batchslot, myslot);
     myslot = batchslot;
    }
   }

   /* ensure that triggers etc see the right relation  */
   myslot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
  }

  skip_tuple = false;

  /* BEFORE ROW INSERT Triggers */
  if (has_before_insert_row_trig)
  {
   if (!ExecBRInsertTriggers(estate, resultRelInfo, myslot))
    skip_tuple = true; /* "do nothing" */
  }

  if (!skip_tuple)
  {
   /*
    * If there is an INSTEAD OF INSERT ROW trigger, let it handle the
    * tuple.  Otherwise, proceed with inserting the tuple into the
    * table or foreign table.
 */
   if (has_instead_insert_row_trig)
   {
    ExecIRInsertTriggers(estate, resultRelInfo, myslot);
   }
   else
   {
    /* Compute stored generated columns */
    if (resultRelInfo->ri_RelationDesc->rd_att->constr &&
     resultRelInfo->ri_RelationDesc->rd_att->constr->has_generated_stored)
     ExecComputeStoredGenerated(resultRelInfo, estate, myslot,
              CMD_INSERT);

    /*
     * If the target is a plain table, check the constraints of
     * the tuple.
 */
    if (resultRelInfo->ri_FdwRoutine == NULL &&
     resultRelInfo->ri_RelationDesc->rd_att->constr)
     ExecConstraints(resultRelInfo, myslot, estate);

    /*
     * Also check the tuple against the partition constraint, if
     * there is one; except that if we got here via tuple-routing,
     * we don't need to if there's no BR trigger defined on the
     * partition.
 */
    if (resultRelInfo->ri_RelationDesc->rd_rel->relispartition &&
     (proute == NULL || has_before_insert_row_trig))
     ExecPartitionCheck(resultRelInfo, myslot, estate, true);

    /* Store the slot in the multi-insert buffer, when enabled. */
    if (insertMethod == CIM_MULTI || leafpart_use_multi_insert)
    {
     /*
      * The slot previously might point into the per-tuple
      * context. For batching it needs to be longer lived.
 */
     ExecMaterializeSlot(myslot);

     /* Add this tuple to the tuple buffer */
     CopyMultiInsertInfoStore(&multiInsertInfo,
            resultRelInfo, myslot,
            cstate->line_buf.len,
            cstate->cur_lineno);

     /*
      * If enough inserts have queued up, then flush all
      * buffers out to their tables.
 */
     if (CopyMultiInsertInfoIsFull(&multiInsertInfo))
      CopyMultiInsertInfoFlush(&multiInsertInfo,
             resultRelInfo,
             &processed);

     /*
      * We delay updating the row counter and progress of the
      * COPY command until after writing the tuples stored in
      * the buffer out to the table, as in single insert mode.
      * See CopyMultiInsertBufferFlush().
 */
     continue; /* next tuple please */
    }
    else
    {
     List    *recheckIndexes = NIL;

     /* OK, store the tuple */
     if (resultRelInfo->ri_FdwRoutine != NULL)
     {
      myslot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
                     resultRelInfo,
                     myslot,
                     NULL);

      if (myslot == NULL) /* "do nothing" */
       continue; /* next tuple please */

      /*
       * AFTER ROW Triggers might reference the tableoid
       * column, so (re-)initialize tts_tableOid before
       * evaluating them.
 */
      myslot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
     }
     else
     {
      /* OK, store the tuple and create index entries for it */
      table_tuple_insert(resultRelInfo->ri_RelationDesc,
             myslot, mycid, ti_options, bistate);

      if (resultRelInfo->ri_NumIndices > 0)
       recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
                   myslot,
                   estate,
                   false,
                   false,
                   NULL,
                   NIL,
                   false);
     }

     /* AFTER ROW INSERT Triggers */
     ExecARInsertTriggers(estate, resultRelInfo, myslot,
           recheckIndexes, cstate->transition_capture);

     list_free(recheckIndexes);
    }
   }

   /*
    * We count only tuples not suppressed by a BEFORE INSERT trigger
    * or FDW; this is the same definition used by nodeModifyTable.c
    * for counting tuples inserted by an INSERT command.  Update
    * progress of the COPY command as well.
 */
   pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
           ++processed);
  }
 }

 /* Flush any remaining buffered tuples */
 if (insertMethod != CIM_SINGLE)
 {
  if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
   CopyMultiInsertInfoFlush(&multiInsertInfo, NULL, &processed);
 }

 /* Done, clean up */
 error_context_stack = errcallback.previous;

 if (cstate->opts.on_error != COPY_ON_ERROR_STOP &&
  cstate->num_errors > 0 &&
  cstate->opts.log_verbosity >= COPY_LOG_VERBOSITY_DEFAULT)
  ereport(NOTICE,
    errmsg_plural("%" PRIu64 " row was skipped due to data type incompatibility",
         "%" PRIu64 " rows were skipped due to data type incompatibility",
         cstate->num_errors,
         cstate->num_errors));

 if (bistate != NULL)
  FreeBulkInsertState(bistate);

 MemoryContextSwitchTo(oldcontext);

 /* Execute AFTER STATEMENT insertion triggers */
 ExecASInsertTriggers(estate, target_resultRelInfo, cstate->transition_capture);

 /* Handle queued AFTER triggers */
 AfterTriggerEndQuery(estate);

 ExecResetTupleTable(estate->es_tupleTable, false);

 /* Allow the FDW to shut down */
 if (target_resultRelInfo->ri_FdwRoutine != NULL &&
  target_resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL)
  target_resultRelInfo->ri_FdwRoutine->EndForeignInsert(estate,
                 target_resultRelInfo);

 /* Tear down the multi-insert buffer data */
 if (insertMethod != CIM_SINGLE)
  CopyMultiInsertInfoCleanup(&multiInsertInfo);

 /* Close all the partitioned tables, leaf partitions, and their indices */
 if (proute)
  ExecCleanupTupleRouting(mtstate, proute);

 /* Close the result relations, including any trigger target relations */
 ExecCloseResultRelations(estate);
 ExecCloseRangeTableRelations(estate);

 FreeExecutorState(estate);

 return processed;
}

/*
 * Setup to read tuples from a file for COPY FROM.
 *
 * 'rel': Used as a template for the tuples
 * 'whereClause': WHERE clause from the COPY FROM command
 * 'filename': Name of server-local file to read, NULL for STDIN
 * 'is_program': true if 'filename' is program to execute
 * 'data_source_cb': callback that provides the input data
 * 'attnamelist': List of char *, columns to include. NIL selects all cols.
 * 'options': List of DefElem. See copy_opt_item in gram.y for selections.
 *
 * Returns a CopyFromState, to be passed to NextCopyFrom and related functions.
 */
CopyFromState
BeginCopyFrom(ParseState *pstate,
     Relation rel,
     Node *whereClause,
     const char *filename,
     bool is_program,
     copy_data_source_cb data_source_cb,
     List *attnamelist,
     List *options)
{
 CopyFromState cstate;
 bool  pipe = (filename == NULL);
 TupleDesc tupDesc;
 AttrNumber num_phys_attrs,
    num_defaults;
 FmgrInfo   *in_functions;
 Oid     *typioparams;
 int     *defmap;
 ExprState **defexprs;
 MemoryContext oldcontext;
 bool  volatile_defexprs;
 const int progress_cols[] = {
  PROGRESS_COPY_COMMAND,
  PROGRESS_COPY_TYPE,
  PROGRESS_COPY_BYTES_TOTAL
 };
 int64  progress_vals[] = {
  PROGRESS_COPY_COMMAND_FROM,
  0,
  0
 };

 /* Allocate workspace and zero all fields */
 cstate = (CopyFromStateData *) palloc0(sizeof(CopyFromStateData));

 /*
  * We allocate everything used by a cstate in a new memory context. This
  * avoids memory leaks during repeated use of COPY in a query.
 */
 cstate->copycontext = AllocSetContextCreate(CurrentMemoryContext,
            "COPY",
            ALLOCSET_DEFAULT_SIZES);

 oldcontext = MemoryContextSwitchTo(cstate->copycontext);

 /* Extract options from the statement node tree */
 ProcessCopyOptions(pstate, &cstate->opts, true /* is_from */ , options);

 /* Set the format routine */
 cstate->routine = CopyFromGetRoutine(&cstate->opts);

 /* Process the target relation */
 cstate->rel = rel;

 tupDesc = RelationGetDescr(cstate->rel);

 /* process common options or initialization */

 /* Generate or convert list of attributes to process */
 cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);

 num_phys_attrs = tupDesc->natts;

 /* Convert FORCE_NOT_NULL name list to per-column flags, check validity */
 cstate->opts.force_notnull_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
 if (cstate->opts.force_notnull_all)
  MemSet(cstate->opts.force_notnull_flags, true, num_phys_attrs * sizeof(bool));
 else if (cstate->opts.force_notnull)
 {
  List    *attnums;
  ListCell   *cur;

  attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_notnull);

  foreach(cur, attnums)
  {
   int   attnum = lfirst_int(cur);
   Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);

   if (!list_member_int(cstate->attnumlist, attnum))
    ereport(ERROR,
      (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
    /*- translator: first %s is the name of a COPY option, e.g. FORCE_NOT_NULL */
       errmsg("%s column \"%s\" not referenced by COPY",
        "FORCE_NOT_NULL", NameStr(attr->attname))));
   cstate->opts.force_notnull_flags[attnum - 1] = true;
  }
 }

 /* Set up soft error handler for ON_ERROR */
 if (cstate->opts.on_error != COPY_ON_ERROR_STOP)
 {
  cstate->escontext = makeNode(ErrorSaveContext);
  cstate->escontext->type = T_ErrorSaveContext;
  cstate->escontext->error_occurred = false;

  /*
   * Currently we only support COPY_ON_ERROR_IGNORE. We'll add other
   * options later
 */
  if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE)
   cstate->escontext->details_wanted = false;
 }
 else
  cstate->escontext = NULL;

 /* Convert FORCE_NULL name list to per-column flags, check validity */
 cstate->opts.force_null_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
 if (cstate->opts.force_null_all)
  MemSet(cstate->opts.force_null_flags, true, num_phys_attrs * sizeof(bool));
 else if (cstate->opts.force_null)
 {
  List    *attnums;
  ListCell   *cur;

  attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_null);

  foreach(cur, attnums)
  {
   int   attnum = lfirst_int(cur);
   Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);

   if (!list_member_int(cstate->attnumlist, attnum))
    ereport(ERROR,
      (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
    /*- translator: first %s is the name of a COPY option, e.g. FORCE_NOT_NULL */
       errmsg("%s column \"%s\" not referenced by COPY",
        "FORCE_NULL", NameStr(attr->attname))));
   cstate->opts.force_null_flags[attnum - 1] = true;
  }
 }

 /* Convert convert_selectively name list to per-column flags */
 if (cstate->opts.convert_selectively)
 {
  List    *attnums;
  ListCell   *cur;

  cstate->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));

  attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.convert_select);

  foreach(cur, attnums)
  {
   int   attnum = lfirst_int(cur);
   Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);

   if (!list_member_int(cstate->attnumlist, attnum))
    ereport(ERROR,
      (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
       errmsg_internal("selected column \"%s\" not referenced by COPY",
           NameStr(attr->attname))));
   cstate->convert_select_flags[attnum - 1] = true;
  }
 }

 /* Use client encoding when ENCODING option is not specified. */
 if (cstate->opts.file_encoding < 0)
  cstate->file_encoding = pg_get_client_encoding();
 else
  cstate->file_encoding = cstate->opts.file_encoding;

 /*
  * Look up encoding conversion function.
 */
 if (cstate->file_encoding == GetDatabaseEncoding() ||
  cstate->file_encoding == PG_SQL_ASCII ||
  GetDatabaseEncoding() == PG_SQL_ASCII)
 {
  cstate->need_transcoding = false;
 }
 else
 {
  cstate->need_transcoding = true;
  cstate->conversion_proc = FindDefaultConversionProc(cstate->file_encoding,
               GetDatabaseEncoding());
  if (!OidIsValid(cstate->conversion_proc))
   ereport(ERROR,
     (errcode(ERRCODE_UNDEFINED_FUNCTION),
      errmsg("default conversion function for encoding \"%s\" to \"%s\" does not exist",
       pg_encoding_to_char(cstate->file_encoding),
       pg_encoding_to_char(GetDatabaseEncoding()))));
 }

 cstate->copy_src = COPY_FILE; /* default */

 cstate->whereClause = whereClause;

 /* Initialize state variables */
 cstate->eol_type = EOL_UNKNOWN;
 cstate->cur_relname = RelationGetRelationName(cstate->rel);
 cstate->cur_lineno = 0;
 cstate->cur_attname = NULL;
 cstate->cur_attval = NULL;
 cstate->relname_only = false;

 /*
  * Allocate buffers for the input pipeline.
  *
  * attribute_buf and raw_buf are used in both text and binary modes, but
  * input_buf and line_buf only in text mode.
 */
 cstate->raw_buf = palloc(RAW_BUF_SIZE + 1);
 cstate->raw_buf_index = cstate->raw_buf_len = 0;
 cstate->raw_reached_eof = false;

 initStringInfo(&cstate->attribute_buf);

 /* Assign range table and rteperminfos, we'll need them in CopyFrom. */
 if (pstate)
 {
  cstate->range_table = pstate->p_rtable;
  cstate->rteperminfos = pstate->p_rteperminfos;
 }

 num_defaults = 0;
 volatile_defexprs = false;

 /*
  * Pick up the required catalog information for each attribute in the
  * relation, including the input function, the element type (to pass to
  * the input function), and info about defaults and constraints. (Which
  * input function we use depends on text/binary format choice.)
 */
 in_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
 typioparams = (Oid *) palloc(num_phys_attrs * sizeof(Oid));
 defmap = (int *) palloc(num_phys_attrs * sizeof(int));
 defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));

 for (int attnum = 1; attnum <= num_phys_attrs; attnum++)
 {
  Form_pg_attribute att = TupleDescAttr(tupDesc, attnum - 1);

  /* We don't need info for dropped attributes */
  if (att->attisdropped)
   continue;

  /* Fetch the input function and typioparam info */
  cstate->routine->CopyFromInFunc(cstate, att->atttypid,
          &in_functions[attnum - 1],
          &typioparams[attnum - 1]);

  /* Get default info if available */
  defexprs[attnum - 1] = NULL;

  /*
   * We only need the default values for columns that do not appear in
   * the column list, unless the DEFAULT option was given. We never need
   * default values for generated columns.
 */
  if ((cstate->opts.default_print != NULL ||
    !list_member_int(cstate->attnumlist, attnum)) &&
   !att->attgenerated)
  {
   Expr    *defexpr = (Expr *) build_column_default(cstate->rel,
                attnum);

   if (defexpr != NULL)
   {
    /* Run the expression through planner */
    defexpr = expression_planner(defexpr);

    /* Initialize executable expression in copycontext */
    defexprs[attnum - 1] = ExecInitExpr(defexpr, NULL);

    /* if NOT copied from input */
    /* use default value if one exists */
    if (!list_member_int(cstate->attnumlist, attnum))
    {
     defmap[num_defaults] = attnum - 1;
     num_defaults++;
    }

    /*
     * If a default expression looks at the table being loaded,
     * then it could give the wrong answer when using
     * multi-insert. Since database access can be dynamic this is
     * hard to test for exactly, so we use the much wider test of
     * whether the default expression is volatile. We allow for
     * the special case of when the default expression is the
     * nextval() of a sequence which in this specific case is
     * known to be safe for use with the multi-insert
     * optimization. Hence we use this special case function
     * checker rather than the standard check for
     * contain_volatile_functions().  Note also that we already
     * ran the expression through expression_planner().
 */
    if (!volatile_defexprs)
     volatile_defexprs = contain_volatile_functions_not_nextval((Node *) defexpr);
   }
  }
 }

 cstate->defaults = (bool *) palloc0(tupDesc->natts * sizeof(bool));

 /* initialize progress */
 pgstat_progress_start_command(PROGRESS_COMMAND_COPY,
          cstate->rel ? RelationGetRelid(cstate->rel) : InvalidOid);
 cstate->bytes_processed = 0;

 /* We keep those variables in cstate. */
 cstate->in_functions = in_functions;
 cstate->typioparams = typioparams;
 cstate->defmap = defmap;
 cstate->defexprs = defexprs;
 cstate->volatile_defexprs = volatile_defexprs;
 cstate->num_defaults = num_defaults;
 cstate->is_program = is_program;

 if (data_source_cb)
 {
  progress_vals[1] = PROGRESS_COPY_TYPE_CALLBACK;
  cstate->copy_src = COPY_CALLBACK;
  cstate->data_source_cb = data_source_cb;
 }
 else if (pipe)
 {
  progress_vals[1] = PROGRESS_COPY_TYPE_PIPE;
  Assert(!is_program); /* the grammar does not allow this */
  if (whereToSendOutput == DestRemote)
   ReceiveCopyBegin(cstate);
  else
   cstate->copy_file = stdin;
 }
 else
 {
  cstate->filename = pstrdup(filename);

  if (cstate->is_program)
  {
   progress_vals[1] = PROGRESS_COPY_TYPE_PROGRAM;
   cstate->copy_file = OpenPipeStream(cstate->filename, PG_BINARY_R);
   if (cstate->copy_file == NULL)
    ereport(ERROR,
      (errcode_for_file_access(),
       errmsg("could not execute command \"%s\": %m",
        cstate->filename)));
  }
  else
  {
   struct stat st;

   progress_vals[1] = PROGRESS_COPY_TYPE_FILE;
   cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_R);
   if (cstate->copy_file == NULL)
   {
    /* copy errno because ereport subfunctions might change it */
    int   save_errno = errno;

    ereport(ERROR,
      (errcode_for_file_access(),
       errmsg("could not open file \"%s\" for reading: %m",
        cstate->filename),
       (save_errno == ENOENT || save_errno == EACCES) ?
       errhint("COPY FROM instructs the PostgreSQL server process to read a file. "
         "You may want a client-side facility such as psql's \\copy.") : 0));
   }

   if (fstat(fileno(cstate->copy_file), &st))
    ereport(ERROR,
      (errcode_for_file_access(),
       errmsg("could not stat file \"%s\": %m",
        cstate->filename)));

   if (S_ISDIR(st.st_mode))
    ereport(ERROR,
      (errcode(ERRCODE_WRONG_OBJECT_TYPE),
       errmsg("\"%s\" is a directory", cstate->filename)));

   progress_vals[2] = st.st_size;
  }
 }

 pgstat_progress_update_multi_param(3, progress_cols, progress_vals);

 cstate->routine->CopyFromStart(cstate, tupDesc);

 MemoryContextSwitchTo(oldcontext);

 return cstate;
}

/*
 * Clean up storage and release resources for COPY FROM.
 */
void
EndCopyFrom(CopyFromState cstate)
{
 /* Invoke the end callback */
 cstate->routine->CopyFromEnd(cstate);

 /* No COPY FROM related resources except memory. */
 if (cstate->is_program)
 {
  ClosePipeFromProgram(cstate);
 }
 else
 {
  if (cstate->filename != NULL && FreeFile(cstate->copy_file))
   ereport(ERROR,
     (errcode_for_file_access(),
      errmsg("could not close file \"%s\": %m",
       cstate->filename)));
 }

 pgstat_progress_end_command();

 MemoryContextDelete(cstate->copycontext);
 pfree(cstate);
}

/*
 * Closes the pipe from an external program, checking the pclose() return code.
 */
static void
ClosePipeFromProgram(CopyFromState cstate)
{
 int   pclose_rc;

 Assert(cstate->is_program);

 pclose_rc = ClosePipeStream(cstate->copy_file);
 if (pclose_rc == -1)
  ereport(ERROR,
    (errcode_for_file_access(),
     errmsg("could not close pipe to external command: %m")));
 else if (pclose_rc != 0)
 {
  /*
   * If we ended a COPY FROM PROGRAM before reaching EOF, then it's
   * expectable for the called program to fail with SIGPIPE, and we
   * should not report that as an error.  Otherwise, SIGPIPE indicates a
   * problem.
 */
  if (!cstate->raw_reached_eof &&
   wait_result_is_signal(pclose_rc, SIGPIPE))
   return;

  ereport(ERROR,
    (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
     errmsg("program \"%s\" failed",
      cstate->filename),
     errdetail_internal("%s", wait_result_to_str(pclose_rc))));
 }
}