/*------------------------------------------------------------------------- * * copyfrom.c * COPY 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 #include #include #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)))); } }