agora inbox for pgsql-hackers@postgresql.org
help / color / mirror / Atom feedFrom: Andres Freund <andres@anarazel.de>
To: David Rowley <david.rowley@2ndquadrant.com>
Cc: Tomas Vondra <tomas.vondra@2ndquadrant.com>
Cc: Surafel Temesgen <surafel3000@gmail.com>
Cc: Alvaro Herrera <alvherre@2ndquadrant.com>
Cc: Adam Berlin <berlin.ab@gmail.com>
Cc: PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Subject: Re: COPY FROM WHEN condition
Date: Tue, 02 Apr 2019 10:52:16 -0700
Message-ID: <87ef6kl2gf.fsf@alap4.lan> (raw)
In-Reply-To: <CAKJS1f9H7qAQZHoowb=cbC8gAx_oxXzd8FgSiHk94_pHycCzRQ@mail.gmail.com>
On 2019-04-03 06:41:49 +1300, David Rowley wrote:
> However, I've ended up not doing it that way as the patch requires
> more than just an array of TupleTableSlots to be stored in the
> ResultRelInfo, it'll pretty much need all of what I have in
> CopyMultiInsertBuffer, which includes line numbers for error
> reporting, a BulkInsertState and a counter to track how many of the
> slots are used. I had thoughts that we could just tag the whole
> CopyMultiInsertBuffer in ResultRelInfo, but that requires moving it
> somewhere a bit more generic. Another thought would be that we have
> something like "void *extra;" in ResultRelInfo that we can just borrow
> for copy.c. It may be interesting to try this out to see if it saves
> much in the way of performance.
Hm, we could just forwad declare struct CopyMultiInsertBuffer in a
header, and only define it in copy.c. That doesn't sound insane to me.
> I've got the patch into a working state now and wrote a bunch of
> comments. I've not really done a thorough read back of the code again.
> I normally wait until the light is coming from a different angle
> before doing that, but there does not seem to be much time to wait for
> that in this case, so here's v2. Performance seems to be about the
> same as what I reported yesterday.
Cool.
> +/*
> + * Multi-Insert handling code for COPY FROM. Inserts are performed in
> + * batches of up to MAX_BUFFERED_TUPLES.
This should probably be moved to the top of the file.
> + * When COPY FROM is used on a partitioned table, we attempt to maintain
> + * only MAX_PARTITION_BUFFERS buffers at once. Buffers that are completely
> + * unused in each batch are removed so that we end up not keeping buffers for
> + * partitions that we might not insert into again.
> + */
> +#define MAX_BUFFERED_TUPLES 1000
> +#define MAX_BUFFERED_BYTES 65535
> +#define MAX_PARTITION_BUFFERS 15
> +/*
> + * CopyMultiInsertBuffer
> + * Stores multi-insert data related to a single relation in CopyFrom.
> + */
> +typedef struct
Please don't create anonymous structs - they can't be forward declared,
and some debugging tools handle them worse than named structs. And it
makes naming CopyMultiInsertBuffer in the comment less necessary.
> +{
> + Oid relid; /* Relation ID this insert data is for */
> + TupleTableSlot **slots; /* Array of MAX_BUFFERED_TUPLES to store
> + * tuples */
> + uint64 *linenos; /* Line # of tuple in copy stream */
It could make sense to allocate those two together, or even as part of
the CopyMultiInsertBuffer itself, to reduce allocator overhead.
> + else
> + {
> + CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) palloc(sizeof(CopyMultiInsertBuffer));
> +
> + /* Non-partitioned table. Just setup the buffer for the table. */
> + buffer->relid = RelationGetRelid(rri->ri_RelationDesc);
> + buffer->slots = palloc0(MAX_BUFFERED_TUPLES * sizeof(TupleTableSlot *));
> + buffer->linenos = palloc(MAX_BUFFERED_TUPLES * sizeof(uint64));
> + buffer->resultRelInfo = rri;
> + buffer->bistate = GetBulkInsertState();
> + buffer->nused = 0;
> + miinfo->multiInsertBufferTab = NULL;
> + miinfo->buffer = buffer;
> + miinfo->nbuffers = 1;
> + }
Can this be moved into a helper function?
> + /*
> + * heap_multi_insert leaks memory, so switch to short-lived memory context
> + * before calling it.
> + */
s/heap_multi_insert/table_multi_insert/
> + /*
> + * If there are any indexes, update them for all the inserted tuples, and
> + * run AFTER ROW INSERT triggers.
> + */
> + if (resultRelInfo->ri_NumIndices > 0)
> + {
> + for (i = 0; i < nBufferedTuples; i++)
> + {
> + List *recheckIndexes;
> +
> + cstate->cur_lineno = buffer->linenos[i];
> + recheckIndexes =
> + ExecInsertIndexTuples(buffer->slots[i], estate, false, NULL,
> + NIL);
> + ExecARInsertTriggers(estate, resultRelInfo,
> + buffer->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))
> + {
> + for (i = 0; i < nBufferedTuples; i++)
> + {
> + cstate->cur_lineno = buffer->linenos[i];
> + ExecARInsertTriggers(estate, resultRelInfo,
> + bufferedSlots[i],
> + NIL, cstate->transition_capture);
> + }
> + }
> + for (i = 0; i < nBufferedTuples; i++)
> + ExecClearTuple(bufferedSlots[i]);
I wonder about combining these loops somehow. But it's probably ok.
> +/*
> + * CopyMultiInsertBuffer_Cleanup
> + * Drop used slots and free member for this buffer. The buffer
> + * must be flushed before cleanup.
> + */
> +static inline void
> +CopyMultiInsertBuffer_Cleanup(CopyMultiInsertBuffer *buffer)
> +{
> + int i;
> +
> + ReleaseBulkInsertStatePin(buffer->bistate);
Shouldn't this FreeBulkInsertState() rather than
ReleaseBulkInsertStatePin()?
> +
> +/*
> + * CopyMultiInsertBuffer_RemoveBuffer
> + * Remove a buffer from being tracked by miinfo
> + */
> +static inline void
> +CopyMultiInsertBuffer_RemoveBuffer(CopyMultiInsertInfo *miinfo,
> + CopyMultiInsertBuffer *buffer)
> +{
> + Oid relid = buffer->relid;
> +
> + CopyMultiInsertBuffer_Cleanup(buffer);
> +
> + hash_search(miinfo->multiInsertBufferTab, (void *) &relid, HASH_REMOVE,
> + NULL);
> + miinfo->nbuffers--;
Aren't we leaking the CopyMultiInsertBuffer itself here?
> +}
> +
> +/*
> + * CopyMultiInsertInfo_Flush
> + * Write out all stored tuples in all buffers out to the tables.
> + *
> + * To save us from ending up with buffers for 1000s of partitions we remove
> + * buffers belonging to partitions that we've seen no tuples for in this batch
> + */
> +static inline void
> +CopyMultiInsertInfo_Flush(CopyMultiInsertInfo *miinfo, CopyState cstate,
> + EState *estate, CommandId mycid, int ti_options)
> +{
> + CopyMultiInsertBuffer *buffer;
> +
> + /*
> + * If just storing buffers for a non-partitioned table, then just flush
> + * that buffer.
> + */
> + if (miinfo->multiInsertBufferTab == NULL)
> + {
> + buffer = miinfo->buffer;
> +
> + CopyMultiInsertInfo_FlushSingleBuffer(buffer, cstate, estate, mycid,
> + ti_options);
> + }
> + else
> + {
> + HASH_SEQ_STATUS status;
> +
> + /*
> + * Otherwise make a pass over the hash table and flush all buffers
> + * that have any tuples stored in them.
> + */
> + hash_seq_init(&status, miinfo->multiInsertBufferTab);
> +
> + while ((buffer = (CopyMultiInsertBuffer *) hash_seq_search(&status)) != NULL)
> + {
> + if (buffer->nused > 0)
> + {
> + /* Flush the buffer if it was used */
> + CopyMultiInsertInfo_FlushSingleBuffer(buffer, cstate, estate,
> + mycid, ti_options);
> + }
> + else
> + {
> + /*
> + * Otherwise just remove it. If we saw no tuples for it this
> + * batch, then likely its best to make way for buffers for
> + * other partitions.
> + */
> + CopyMultiInsertBuffer_RemoveBuffer(miinfo, buffer);
> + }
> + }
> + }
> +
> + miinfo->bufferedTuples = 0;
> + miinfo->bufferedBytes = 0;
> +}
> +
> +/*
> + * CopyMultiInsertInfo_Cleanup
> + * Cleanup allocated buffers and free memory
> + */
> +static inline void
> +CopyMultiInsertInfo_Cleanup(CopyMultiInsertInfo *miinfo)
> +{
> + if (miinfo->multiInsertBufferTab == NULL)
> + CopyMultiInsertBuffer_Cleanup(miinfo->buffer);
> + else
> + {
> + HASH_SEQ_STATUS status;
> + CopyMultiInsertBuffer *buffer;
> +
> + hash_seq_init(&status, miinfo->multiInsertBufferTab);
> +
> + while ((buffer = (CopyMultiInsertBuffer *) hash_seq_search(&status)) != NULL)
> + {
> + Assert(buffer->nused == 0);
> + CopyMultiInsertBuffer_Cleanup(buffer);
> + }
> +
> + hash_destroy(miinfo->multiInsertBufferTab);
> + }
> +}
> +
> +/*
> + * CopyMultiInsertInfo_NextFreeSlot
> + * Get the next TupleTableSlot that the next tuple should be stored in.
> + *
> + * Callers must ensure that the buffer is not full.
> + */
> +static inline TupleTableSlot *
> +CopyMultiInsertInfo_NextFreeSlot(CopyMultiInsertInfo *miinfo,
> + ResultRelInfo *rri)
> +{
> + CopyMultiInsertBuffer *buffer = miinfo->buffer;
> + int nused = buffer->nused;
> +
> + Assert(nused < MAX_BUFFERED_TUPLES);
> +
> + if (buffer->slots[nused] == NULL)
> + buffer->slots[nused] = table_slot_create(rri->ri_RelationDesc, NULL);
> + return buffer->slots[nused];
> +}
> +
> +/*
> + * CopyMultiInsertInfo_Store
> + * Consume the previously reserved TupleTableSlot that was reserved by
> + * CopyMultiInsertInfo_NextFreeSlot.
> + */
> +static inline void
> +CopyMultiInsertInfo_Store(CopyMultiInsertInfo *miinfo, TupleTableSlot *slot,
> + int tuplen, uint64 lineno)
> +{
> + CopyMultiInsertBuffer *buffer = miinfo->buffer;
> +
> + 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(CopyState cstate)
> {
> - HeapTuple tuple;
> - TupleDesc tupDesc;
> - Datum *values;
> - bool *nulls;
> ResultRelInfo *resultRelInfo;
> ResultRelInfo *target_resultRelInfo;
> ResultRelInfo *prevResultRelInfo = NULL;
> EState *estate = CreateExecutorState(); /* for ExecConstraints() */
> ModifyTableState *mtstate;
> ExprContext *econtext;
> - TupleTableSlot *myslot;
> + TupleTableSlot *singleslot = NULL;
> MemoryContext oldcontext = CurrentMemoryContext;
> - MemoryContext batchcontext;
>
> PartitionTupleRouting *proute = NULL;
> ErrorContextCallback errcallback;
> CommandId mycid = GetCurrentCommandId(true);
> int ti_options = 0; /* start with default table_insert options */
> - BulkInsertState bistate;
> + BulkInsertState bistate = NULL;
> CopyInsertMethod insertMethod;
> + CopyMultiInsertInfo multiInsertInfo;
> uint64 processed = 0;
> - int nBufferedTuples = 0;
> bool has_before_insert_row_trig;
> bool has_instead_insert_row_trig;
> bool leafpart_use_multi_insert = false;
>
> -#define MAX_BUFFERED_TUPLES 1000
> -#define RECHECK_MULTI_INSERT_THRESHOLD 1000
> - HeapTuple *bufferedTuples = NULL; /* initialize to silence warning */
> - Size bufferedTuplesSize = 0;
> - uint64 firstBufferedLineNo = 0;
> - uint64 lastPartitionSampleLineNo = 0;
> - uint64 nPartitionChanges = 0;
> - double avgTuplesPerPartChange = 0;
> -
> Assert(cstate->rel);
>
> + memset(&multiInsertInfo, 0, sizeof(CopyMultiInsertInfo));
> +
> /*
> * The target must be a plain, foreign, or partitioned relation, or have
> * an INSTEAD OF INSERT row trigger. (Currently, such triggers are only
> @@ -2382,8 +2767,6 @@ CopyFrom(CopyState cstate)
> RelationGetRelationName(cstate->rel))));
> }
>
> - tupDesc = RelationGetDescr(cstate->rel);
> -
> /*----------
> * Check to see if we can avoid writing WAL
> *
> @@ -2467,8 +2850,8 @@ CopyFrom(CopyState cstate)
> if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
> {
> ereport(ERROR,
> - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
> - errmsg("cannot perform FREEZE on a partitioned table")));
> + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
> + errmsg("cannot perform FREEZE on a partitioned table")));
> }
>
> /*
> @@ -2518,10 +2901,6 @@ CopyFrom(CopyState cstate)
>
> ExecInitRangeTable(estate, cstate->range_table);
>
> - /* Set up a tuple slot too */
> - myslot = ExecInitExtraTupleSlot(estate, tupDesc,
> - &TTSOpsHeapTuple);
> -
> /*
> * Set up a ModifyTableState so we can let FDW(s) init themselves for
> * foreign-table result relation(s).
> @@ -2565,10 +2944,11 @@ CopyFrom(CopyState cstate)
> &mtstate->ps);
>
> /*
> - * It's more efficient to prepare a bunch of tuples for insertion, and
> - * insert them in one heap_multi_insert() call, than call heap_insert()
> - * separately for every tuple. However, there are a number of reasons why
> - * we might not be able to do this. These are explained below.
> + * It's generally more efficient to prepare a bunch of tuples for
> + * insertion, and insert them in one table_multi_insert() call, than call
> + * table_insert() 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 ||
> @@ -2589,8 +2969,8 @@ CopyFrom(CopyState cstate)
> * 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, CopyFromInsertBatch expects that any before row insert and
> - * statement level insert triggers are on the same relation.
> + * now, CopyMultiInsertInfo_Flush expects that any before row insert
> + * and statement level insert triggers are on the same relation.
> */
> insertMethod = CIM_SINGLE;
> }
> @@ -2622,8 +3002,7 @@ CopyFrom(CopyState cstate)
> {
> /*
> * For partitioned tables, we may still be able to perform bulk
> - * inserts for sets of consecutive tuples which belong to the same
> - * partition. However, the possibility of this depends on which types
> + * 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 or it has any before row insert
> * or insert instead triggers (same as we checked above for the parent
> @@ -2632,18 +3011,27 @@ CopyFrom(CopyState cstate)
> * 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.
> - *
> - * Normally, when performing bulk inserts we just flush the insert
> - * buffer whenever it becomes full, but for the partitioned table
> - * case, we flush it whenever the current tuple does not belong to the
> - * same partition as the previous tuple.
> */
> if (proute)
> insertMethod = CIM_MULTI_CONDITIONAL;
> else
> insertMethod = CIM_MULTI;
>
> - bufferedTuples = palloc(MAX_BUFFERED_TUPLES * sizeof(HeapTuple));
> + CopyMultiInsertInfo_Init(&multiInsertInfo, resultRelInfo,
> + proute != NULL);
> + }
> +
> + /*
> + * 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 &&
> @@ -2660,10 +3048,6 @@ CopyFrom(CopyState cstate)
> */
> ExecBSInsertTriggers(estate, resultRelInfo);
>
> - values = (Datum *) palloc(tupDesc->natts * sizeof(Datum));
> - nulls = (bool *) palloc(tupDesc->natts * sizeof(bool));
> -
> - bistate = GetBulkInsertState();
> econtext = GetPerTupleExprContext(estate);
>
> /* Set up callback to identify error line number */
> @@ -2672,17 +3056,9 @@ CopyFrom(CopyState cstate)
> errcallback.previous = error_context_stack;
> error_context_stack = &errcallback;
>
> - /*
> - * Set up memory context for batches. For cases without batching we could
> - * use the per-tuple context, but it's simpler to just use it every time.
> - */
> - batchcontext = AllocSetContextCreate(CurrentMemoryContext,
> - "batch context",
> - ALLOCSET_DEFAULT_SIZES);
> -
> for (;;)
> {
> - TupleTableSlot *slot;
> + TupleTableSlot *myslot;
> bool skip_tuple;
>
> CHECK_FOR_INTERRUPTS();
> @@ -2693,20 +3069,33 @@ CopyFrom(CopyState cstate)
> */
> ResetPerTupleExprContext(estate);
>
> + if (insertMethod == CIM_SINGLE || proute)
> + {
> + myslot = singleslot;
> + Assert(myslot != NULL);
> + }
> + else
> + {
> + Assert(resultRelInfo == target_resultRelInfo);
> + Assert(insertMethod == CIM_MULTI);
> +
> + myslot = CopyMultiInsertInfo_NextFreeSlot(&multiInsertInfo,
> + resultRelInfo);
> + }
> +
> /*
> * Switch to per-tuple context before calling NextCopyFrom, which does
> * evaluate default expressions etc. and requires per-tuple context.
> */
> MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
>
> - if (!NextCopyFrom(cstate, econtext, values, nulls))
> - break;
> + ExecClearTuple(myslot);
>
> - /* Switch into per-batch memory context before forming the tuple. */
> - MemoryContextSwitchTo(batchcontext);
> + /* Directly store the values/nulls array in the slot */
> + if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
> + break;
>
> - /* And now we can form the input tuple. */
> - tuple = heap_form_tuple(tupDesc, values, nulls);
> + ExecStoreVirtualTuple(myslot);
>
> /*
> * Constraints might reference the tableoid column, so (re-)initialize
> @@ -2717,18 +3106,15 @@ CopyFrom(CopyState cstate)
> /* Triggers and stuff need to be invoked in query context. */
> MemoryContextSwitchTo(oldcontext);
>
> - /* Place tuple in tuple slot --- but slot shouldn't free it */
> - slot = myslot;
> - ExecStoreHeapTuple(tuple, slot, false);
> -
> if (cstate->whereClause)
> {
> econtext->ecxt_scantuple = myslot;
> + /* Skip items that don't match the COPY's WHERE clause */
> if (!ExecQual(cstate->qualexpr, econtext))
> continue;
> }
>
> - /* Determine the partition to heap_insert the tuple into */
> + /* Determine the partition to table_insert the tuple into */
> if (proute)
> {
> TupleConversionMap *map;
> @@ -2739,80 +3125,10 @@ CopyFrom(CopyState cstate)
> * if the found partition is not suitable for INSERTs.
> */
> resultRelInfo = ExecFindPartition(mtstate, target_resultRelInfo,
> - proute, slot, estate);
> + proute, myslot, estate);
>
> if (prevResultRelInfo != resultRelInfo)
> {
> - /* Check if we can multi-insert into this partition */
> - if (insertMethod == CIM_MULTI_CONDITIONAL)
> - {
> - /*
> - * When performing bulk-inserts into partitioned tables we
> - * must insert the tuples seen so far to the heap whenever
> - * the partition changes.
> - */
> - if (nBufferedTuples > 0)
> - {
> - MemoryContext oldcontext;
> -
> - CopyFromInsertBatch(cstate, estate, mycid, ti_options,
> - prevResultRelInfo, myslot, bistate,
> - nBufferedTuples, bufferedTuples,
> - firstBufferedLineNo);
> - nBufferedTuples = 0;
> - bufferedTuplesSize = 0;
> -
> - /*
> - * The tuple is already allocated in the batch context, which
> - * we want to reset. So to keep the tuple we copy it into the
> - * short-lived (per-tuple) context, reset the batch context
> - * and then copy it back into the per-batch one.
> - */
> - oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
> - tuple = heap_copytuple(tuple);
> - MemoryContextSwitchTo(oldcontext);
> -
> - /* cleanup the old batch */
> - MemoryContextReset(batchcontext);
> -
> - /* copy the tuple back to the per-batch context */
> - oldcontext = MemoryContextSwitchTo(batchcontext);
> - tuple = heap_copytuple(tuple);
> - MemoryContextSwitchTo(oldcontext);
> -
> - /*
> - * Also push the tuple copy to the slot (resetting the context
> - * invalidated the slot contents).
> - */
> - ExecStoreHeapTuple(tuple, slot, false);
> - }
> -
> - nPartitionChanges++;
> -
> - /*
> - * Here we adaptively enable multi-inserts based on the
> - * average number of tuples from recent multi-insert
> - * batches. We recalculate the average every
> - * RECHECK_MULTI_INSERT_THRESHOLD tuples instead of taking
> - * the average over the whole copy. This allows us to
> - * enable multi-inserts when we get periods in the copy
> - * stream that have tuples commonly belonging to the same
> - * partition, and disable when the partition is changing
> - * too often.
> - */
> - if (unlikely(lastPartitionSampleLineNo <= (cstate->cur_lineno -
> - RECHECK_MULTI_INSERT_THRESHOLD)
> - && cstate->cur_lineno >= RECHECK_MULTI_INSERT_THRESHOLD))
> - {
> - avgTuplesPerPartChange =
> - (cstate->cur_lineno - lastPartitionSampleLineNo) /
> - (double) nPartitionChanges;
> -
> - lastPartitionSampleLineNo = cstate->cur_lineno;
> - nPartitionChanges = 0;
> - }
> - }
> -
> /* Determine which triggers exist on this partition */
> has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
> resultRelInfo->ri_TrigDesc->trig_insert_before_row);
> @@ -2821,22 +3137,19 @@ CopyFrom(CopyState cstate)
> resultRelInfo->ri_TrigDesc->trig_insert_instead_row);
>
> /*
> - * Tests have shown that using multi-inserts when the
> - * partition changes on every tuple slightly decreases the
> - * performance, however, there are benefits even when only
> - * some batches have just 2 tuples, so let's enable
> - * multi-inserts even when the average is quite low.
> + * Enable multi-inserts when the partition has BEFORE/INSTEAD
> + * OF triggers, or if the partition is a foreign partition.
> */
> leafpart_use_multi_insert = insertMethod == CIM_MULTI_CONDITIONAL &&
> - avgTuplesPerPartChange >= 1.3 &&
> !has_before_insert_row_trig &&
> !has_instead_insert_row_trig &&
> resultRelInfo->ri_FdwRoutine == NULL;
>
> - /*
> - * We'd better make the bulk insert mechanism gets a new
> - * buffer when the partition being inserted into changes.
> - */
> + /* Set the multi-insert buffer to use for this partition. */
> + if (leafpart_use_multi_insert)
> + CopyMultiInsertInfo_SetCurrentBuffer(&multiInsertInfo,
> + resultRelInfo);
> +
> ReleaseBulkInsertStatePin(bistate);
> prevResultRelInfo = resultRelInfo;
> }
> @@ -2879,26 +3192,48 @@ CopyFrom(CopyState cstate)
> * rowtype.
> */
> map = resultRelInfo->ri_PartitionInfo->pi_RootToPartitionMap;
> - if (map != NULL)
> + if (insertMethod == CIM_SINGLE || !leafpart_use_multi_insert)
> {
> - TupleTableSlot *new_slot;
> - MemoryContext oldcontext;
> -
> - new_slot = resultRelInfo->ri_PartitionInfo->pi_PartitionTupleSlot;
> - Assert(new_slot != NULL);
> -
> - slot = execute_attr_map_slot(map->attrMap, slot, new_slot);
> + /* non batch insert */
> + if (map != NULL)
> + {
> + TupleTableSlot *new_slot;
>
> + new_slot = resultRelInfo->ri_PartitionInfo->pi_PartitionTupleSlot;
> + myslot = execute_attr_map_slot(map->attrMap, myslot, new_slot);
> + }
> + }
> + else
> + {
> /*
> - * Get the tuple in the per-batch context, so that it will be
> - * freed after each batch insert.
> + * Batch insert into partitioned table.
> */
> - oldcontext = MemoryContextSwitchTo(batchcontext);
> - tuple = ExecCopySlotHeapTuple(slot);
> - MemoryContextSwitchTo(oldcontext);
> + TupleTableSlot *nextslot;
> +
> + /* no other path available for partitioned table */
> + Assert(insertMethod == CIM_MULTI_CONDITIONAL);
> +
> + nextslot = CopyMultiInsertInfo_NextFreeSlot(&multiInsertInfo,
> + resultRelInfo);
> +
> + if (map != NULL)
> + myslot = execute_attr_map_slot(map->attrMap, myslot, nextslot);
> + 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(nextslot, myslot);
> + myslot = nextslot;
> + }
> }
>
> - slot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
> + /* ensure that triggers etc see the right relation */
> + myslot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
> }
>
> skip_tuple = false;
> @@ -2906,7 +3241,7 @@ CopyFrom(CopyState cstate)
> /* BEFORE ROW INSERT Triggers */
> if (has_before_insert_row_trig)
> {
> - if (!ExecBRInsertTriggers(estate, resultRelInfo, slot))
> + if (!ExecBRInsertTriggers(estate, resultRelInfo, myslot))
> skip_tuple = true; /* "do nothing" */
> }
>
> @@ -2919,7 +3254,7 @@ CopyFrom(CopyState cstate)
> */
> if (has_instead_insert_row_trig)
> {
> - ExecIRInsertTriggers(estate, resultRelInfo, slot);
> + ExecIRInsertTriggers(estate, resultRelInfo, myslot);
> }
> else
> {
> @@ -2931,12 +3266,7 @@ CopyFrom(CopyState cstate)
> */
> if (resultRelInfo->ri_RelationDesc->rd_att->constr &&
> resultRelInfo->ri_RelationDesc->rd_att->constr->has_generated_stored)
> - {
> - ExecComputeStoredGenerated(estate, slot);
> - MemoryContextSwitchTo(batchcontext);
> - tuple = ExecCopySlotHeapTuple(slot);
> - MemoryContextSwitchTo(oldcontext);
> - }
> + ExecComputeStoredGenerated(estate, myslot);
>
> /*
> * If the target is a plain table, check the constraints of
> @@ -2944,7 +3274,7 @@ CopyFrom(CopyState cstate)
> */
> if (resultRelInfo->ri_FdwRoutine == NULL &&
> resultRelInfo->ri_RelationDesc->rd_att->constr)
> - ExecConstraints(resultRelInfo, slot, estate);
> + ExecConstraints(resultRelInfo, myslot, estate);
>
> /*
> * Also check the tuple against the partition constraint, if
> @@ -2954,7 +3284,7 @@ CopyFrom(CopyState cstate)
> */
> if (resultRelInfo->ri_PartitionCheck &&
> (proute == NULL || has_before_insert_row_trig))
> - ExecPartitionCheck(resultRelInfo, slot, estate, true);
> + ExecPartitionCheck(resultRelInfo, myslot, estate, true);
>
> /*
> * Perform multi-inserts when enabled, or when loading a
> @@ -2963,31 +3293,21 @@ CopyFrom(CopyState cstate)
> */
> if (insertMethod == CIM_MULTI || leafpart_use_multi_insert)
> {
> - /* Add this tuple to the tuple buffer */
> - if (nBufferedTuples == 0)
> - firstBufferedLineNo = cstate->cur_lineno;
> - bufferedTuples[nBufferedTuples++] = tuple;
> - bufferedTuplesSize += tuple->t_len;
> -
> /*
> - * If the buffer filled up, flush it. Also flush if the
> - * total size of all the tuples in the buffer becomes
> - * large, to avoid using large amounts of memory for the
> - * buffer when the tuples are exceptionally wide.
> + * The slot previously might point into the per-tuple
> + * context. For batching it needs to be longer lived.
> */
> - if (nBufferedTuples == MAX_BUFFERED_TUPLES ||
> - bufferedTuplesSize > 65535)
> - {
> - CopyFromInsertBatch(cstate, estate, mycid, ti_options,
> - resultRelInfo, myslot, bistate,
> - nBufferedTuples, bufferedTuples,
> - firstBufferedLineNo);
> - nBufferedTuples = 0;
> - bufferedTuplesSize = 0;
> -
> - /* free memory occupied by tuples from the batch */
> - MemoryContextReset(batchcontext);
> - }
> + ExecMaterializeSlot(myslot);
> +
> + /* Add this tuple to the tuple buffer */
> + CopyMultiInsertInfo_Store(&multiInsertInfo, myslot,
> + cstate->line_buf.len,
> + cstate->cur_lineno);
> +
> + /* If the buffer filled up, flush it. */
> + if (CopyMultiInsertInfo_IsFull(&multiInsertInfo))
> + CopyMultiInsertInfo_Flush(&multiInsertInfo, cstate,
> + estate, mycid, ti_options);
> }
> else
> {
> @@ -2996,12 +3316,12 @@ CopyFrom(CopyState cstate)
> /* OK, store the tuple */
> if (resultRelInfo->ri_FdwRoutine != NULL)
> {
> - slot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
> - resultRelInfo,
> - slot,
> - NULL);
> + myslot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
> + resultRelInfo,
> + myslot,
> + NULL);
>
> - if (slot == NULL) /* "do nothing" */
> + if (myslot == NULL) /* "do nothing" */
> continue; /* next tuple please */
>
> /*
> @@ -3009,27 +3329,26 @@ CopyFrom(CopyState cstate)
> * column, so (re-)initialize tts_tableOid before
> * evaluating them.
> */
> - slot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
> + myslot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
> }
> else
> {
> - tuple = ExecFetchSlotHeapTuple(slot, true, NULL);
> - heap_insert(resultRelInfo->ri_RelationDesc, tuple,
> - mycid, ti_options, bistate);
> - ItemPointerCopy(&tuple->t_self, &slot->tts_tid);
> - slot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
> + /* OK, store the tuple and create index entries for it */
> + table_insert(resultRelInfo->ri_RelationDesc, myslot,
> + mycid, ti_options, bistate);
> }
>
> +
> /* And create index entries for it */
> if (resultRelInfo->ri_NumIndices > 0)
> - recheckIndexes = ExecInsertIndexTuples(slot,
> + recheckIndexes = ExecInsertIndexTuples(myslot,
> estate,
> false,
> NULL,
> NIL);
>
> /* AFTER ROW INSERT Triggers */
> - ExecARInsertTriggers(estate, resultRelInfo, slot,
> + ExecARInsertTriggers(estate, resultRelInfo, myslot,
> recheckIndexes, cstate->transition_capture);
>
> list_free(recheckIndexes);
> @@ -3045,32 +3364,25 @@ CopyFrom(CopyState cstate)
> }
> }
>
> - /* Flush any remaining buffered tuples */
> - if (nBufferedTuples > 0)
> + if (insertMethod != CIM_SINGLE)
> {
> - if (insertMethod == CIM_MULTI_CONDITIONAL)
> - {
> - CopyFromInsertBatch(cstate, estate, mycid, ti_options,
> - prevResultRelInfo, myslot, bistate,
> - nBufferedTuples, bufferedTuples,
> - firstBufferedLineNo);
> - }
> - else
> - CopyFromInsertBatch(cstate, estate, mycid, ti_options,
> - resultRelInfo, myslot, bistate,
> - nBufferedTuples, bufferedTuples,
> - firstBufferedLineNo);
> + /* Flush any remaining buffered tuples */
> + if (!CopyMultiInsertInfo_IsEmpty(&multiInsertInfo))
> + CopyMultiInsertInfo_Flush(&multiInsertInfo, cstate, estate, mycid,
> + ti_options);
> +
> + /* Tear down the multi-insert data */
> + CopyMultiInsertInfo_Cleanup(&multiInsertInfo);
> }
>
> /* Done, clean up */
> error_context_stack = errcallback.previous;
>
> - FreeBulkInsertState(bistate);
> + if (bistate != NULL)
> + ReleaseBulkInsertStatePin(bistate);
>
> MemoryContextSwitchTo(oldcontext);
>
> - MemoryContextDelete(batchcontext);
> -
> /*
> * In the old protocol, tell pqcomm that we can process normal protocol
> * messages again.
> @@ -3084,9 +3396,6 @@ CopyFrom(CopyState cstate)
> /* Handle queued AFTER triggers */
> AfterTriggerEndQuery(estate);
>
> - pfree(values);
> - pfree(nulls);
> -
> ExecResetTupleTable(estate->es_tupleTable, false);
>
> /* Allow the FDW to shut down */
> @@ -3111,88 +3420,6 @@ CopyFrom(CopyState cstate)
> return processed;
> }
>
> -/*
> - * A subroutine of CopyFrom, to write the current batch of buffered heap
> - * tuples to the heap. Also updates indexes and runs AFTER ROW INSERT
> - * triggers.
> - */
> -static void
> -CopyFromInsertBatch(CopyState cstate, EState *estate, CommandId mycid,
> - int ti_options, ResultRelInfo *resultRelInfo,
> - TupleTableSlot *myslot, BulkInsertState bistate,
> - int nBufferedTuples, HeapTuple *bufferedTuples,
> - uint64 firstBufferedLineNo)
> -{
> - MemoryContext oldcontext;
> - int i;
> - uint64 save_cur_lineno;
> - bool line_buf_valid = cstate->line_buf_valid;
> -
> - /*
> - * Print error context information correctly, if one of the operations
> - * below fail.
> - */
> - cstate->line_buf_valid = false;
> - save_cur_lineno = cstate->cur_lineno;
> -
> - /*
> - * heap_multi_insert leaks memory, so switch to short-lived memory context
> - * before calling it.
> - */
> - oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
> - heap_multi_insert(resultRelInfo->ri_RelationDesc,
> - bufferedTuples,
> - nBufferedTuples,
> - mycid,
> - ti_options,
> - bistate);
> - MemoryContextSwitchTo(oldcontext);
> -
> - /*
> - * If there are any indexes, update them for all the inserted tuples, and
> - * run AFTER ROW INSERT triggers.
> - */
> - if (resultRelInfo->ri_NumIndices > 0)
> - {
> - for (i = 0; i < nBufferedTuples; i++)
> - {
> - List *recheckIndexes;
> -
> - cstate->cur_lineno = firstBufferedLineNo + i;
> - ExecStoreHeapTuple(bufferedTuples[i], myslot, false);
> - recheckIndexes =
> - ExecInsertIndexTuples(myslot,
> - estate, false, NULL, NIL);
> - ExecARInsertTriggers(estate, resultRelInfo,
> - myslot,
> - 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))
> - {
> - for (i = 0; i < nBufferedTuples; i++)
> - {
> - cstate->cur_lineno = firstBufferedLineNo + i;
> - ExecStoreHeapTuple(bufferedTuples[i], myslot, false);
> - ExecARInsertTriggers(estate, resultRelInfo,
> - myslot,
> - NIL, cstate->transition_capture);
> - }
> - }
> -
> - /* reset cur_lineno and line_buf_valid to what they were */
> - cstate->line_buf_valid = line_buf_valid;
> - cstate->cur_lineno = save_cur_lineno;
> -}
> -
> /*
> * Setup to read tuples from a file for COPY FROM.
> *
> @@ -4990,11 +5217,8 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
> DR_copy *myState = (DR_copy *) self;
> CopyState cstate = myState->cstate;
>
> - /* Make sure the tuple is fully deconstructed */
> - slot_getallattrs(slot);
> -
> - /* And send the data */
> - CopyOneRowTo(cstate, slot->tts_values, slot->tts_isnull);
> + /* Send the data */
> + CopyOneRowTo(cstate, slot);
> myState->processed++;
>
> return true;
> diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
> index 4c077755d5..ed0e2de144 100644
> --- a/src/include/access/heapam.h
> +++ b/src/include/access/heapam.h
> @@ -36,6 +36,7 @@
> #define HEAP_INSERT_SPECULATIVE 0x0010
>
> typedef struct BulkInsertStateData *BulkInsertState;
> +struct TupleTableSlot;
>
> #define MaxLockTupleMode LockTupleExclusive
>
> @@ -143,7 +144,7 @@ extern void ReleaseBulkInsertStatePin(BulkInsertState bistate);
>
> extern void heap_insert(Relation relation, HeapTuple tup, CommandId cid,
> int options, BulkInsertState bistate);
> -extern void heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples,
> +extern void heap_multi_insert(Relation relation, struct TupleTableSlot **slots, int ntuples,
> CommandId cid, int options, BulkInsertState bistate);
> extern TM_Result heap_delete(Relation relation, ItemPointer tid,
> CommandId cid, Snapshot crosscheck, bool wait,
> diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
> index 4efe178ed1..c2fdedc551 100644
> --- a/src/include/access/tableam.h
> +++ b/src/include/access/tableam.h
> @@ -328,6 +328,9 @@ typedef struct TableAmRoutine
> * ------------------------------------------------------------------------
> */
>
> + void (*multi_insert) (Relation rel, TupleTableSlot **slots, int nslots,
> + CommandId cid, int options, struct BulkInsertStateData *bistate);
> +
> /* see table_insert() for reference about parameters */
> void (*tuple_insert) (Relation rel, TupleTableSlot *slot,
> CommandId cid, int options,
> @@ -1157,6 +1160,17 @@ table_update(Relation rel, ItemPointer otid, TupleTableSlot *slot,
> lockmode, update_indexes);
> }
>
> +/*
> + * table_multi_insert - insert multiple tuple into a table
> + */
> +static inline void
> +table_multi_insert(Relation rel, TupleTableSlot **slots, int nslots,
> + CommandId cid, int options, struct BulkInsertStateData *bistate)
> +{
> + rel->rd_tableam->multi_insert(rel, slots, nslots,
> + cid, options, bistate);
> +}
> +
> /*
> * Lock a tuple in the specified mode.
> *
Greetings,
Andres Freund
view thread (90+ messages) latest in thread
Message-ID: <87ef6kl2gf.fsf@alap4.lan>
Permalink: ../87ef6kl2gf.fsf@alap4.lan/
Also on: postgresql.org/message-id/87ef6kl2gf.fsf@alap4.lan
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: andres@anarazel.de, david.rowley@2ndquadrant.com, tomas.vondra@2ndquadrant.com, surafel3000@gmail.com, alvherre@2ndquadrant.com, berlin.ab@gmail.com, pgsql-hackers@lists.postgresql.org
Subject: Re: COPY FROM WHEN condition
In-Reply-To: <87ef6kl2gf.fsf@alap4.lan>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox