public inbox for [email protected]
help / color / mirror / Atom feed[PATCH v5 2/2] Make INSERT SELECT use multi_insert
4+ messages / 2 participants
[nested] [flat]
* [PATCH v5 2/2] Make INSERT SELECT use multi_insert
@ 2020-10-25 03:49 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 4+ messages in thread
From: Justin Pryzby @ 2020-10-25 03:49 UTC (permalink / raw)
TODO: CTAS and matview ?
See also: 86b85044e823a304d2a265abc030254d39efe7df
---
src/backend/commands/copy.c | 185 +------------------------
src/backend/executor/nodeModifyTable.c | 167 +++++++++++++++++++---
src/include/commands/copy.h | 185 +++++++++++++++++++++++++
src/include/nodes/execnodes.h | 9 +-
src/test/regress/expected/insert.out | 16 +++
src/test/regress/sql/insert.sql | 5 +
6 files changed, 358 insertions(+), 209 deletions(-)
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 36ddcdccdb..b613e6e43d 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -31,7 +31,6 @@
#include "commands/defrem.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"
@@ -241,54 +240,6 @@ typedef struct
} DR_copy;
-/*
- * 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 */
-#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 */
- 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 */
- CopyState 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;
-
-
/*
* These macros centralize code used to process line_buf and raw_buf buffers.
* They are macros because they often do continue/break control and to avoid
@@ -2386,48 +2337,13 @@ limit_printout_length(const char *str)
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 = GetBulkInsertState();
- 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
+void
CopyMultiInsertInfoInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
CopyState cstate, EState *estate, CommandId mycid,
int ti_options)
@@ -2449,27 +2365,6 @@ CopyMultiInsertInfoInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
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.
*/
@@ -2554,35 +2449,6 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo,
cstate->cur_lineno = save_cur_lineno;
}
-/*
- * Drop used slots and free member for this buffer.
- *
- * The buffer must be flushed before cleanup.
- */
-static inline void
-CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
- CopyMultiInsertBuffer *buffer)
-{
- int i;
-
- /* Ensure buffer was flushed */
- Assert(buffer->nused == 0);
-
- /* Remove back-link to ourself */
- buffer->resultRelInfo->ri_CopyMultiInsertBuffer = NULL;
-
- FreeBulkInsertState(buffer->bistate);
-
- /* 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]);
-
- table_finish_bulk_insert(buffer->resultRelInfo->ri_RelationDesc,
- miinfo->ti_options);
-
- pfree(buffer);
-}
-
/*
* Write out all stored tuples in all buffers out to the tables.
*
@@ -2639,7 +2505,7 @@ CopyMultiInsertInfoFlush(CopyMultiInsertInfo *miinfo, ResultRelInfo *curr_rri)
/*
* Cleanup allocated buffers and free memory
*/
-static inline void
+void
CopyMultiInsertInfoCleanup(CopyMultiInsertInfo *miinfo)
{
ListCell *lc;
@@ -2650,53 +2516,6 @@ CopyMultiInsertInfoCleanup(CopyMultiInsertInfo *miinfo)
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 = buffer->nused;
-
- Assert(buffer != NULL);
- Assert(nused < MAX_BUFFERED_TUPLES);
-
- 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.
*/
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 26ff964105..1437acfc22 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -43,6 +43,7 @@
#include "access/xact.h"
#include "catalog/catalog.h"
#include "commands/trigger.h"
+#include "commands/copy.h"
#include "executor/execPartition.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
@@ -362,6 +363,65 @@ ExecComputeStoredGenerated(ResultRelInfo *resultRelInfo,
MemoryContextSwitchTo(oldContext);
}
+/*
+ * This is copied from CopyMultiInsertInfoFlush and modified to avoid
+ * dragging in CopyState..
+ */
+static void
+InsertMultiInsertInfoFlush(CopyMultiInsertInfo *miinfo)
+{
+ MemoryContext oldcontext;
+ ModifyTableState *mtstate = (ModifyTableState*)miinfo->cstate;
+
+ ListCell *lc;
+ foreach(lc, miinfo->multiInsertBuffers)
+ {
+ CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) lfirst(lc);
+ ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
+
+ oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(miinfo->estate));
+ table_multi_insert(resultRelInfo->ri_RelationDesc,
+ buffer->slots,
+ buffer->nused,
+ miinfo->mycid,
+ miinfo->ti_options,
+ buffer->bistate);
+ MemoryContextSwitchTo(oldcontext);
+
+ for (int i = 0; i < buffer->nused; ++i)
+ {
+ if (resultRelInfo->ri_NumIndices > 0)
+ {
+ List *recheckIndexes;
+ recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
+ buffer->slots[i], miinfo->estate,
+ false, NULL, NIL);
+ ExecARInsertTriggers(miinfo->estate, resultRelInfo,
+ buffer->slots[i], recheckIndexes,
+ mtstate->mt_transition_capture);
+ list_free(recheckIndexes);
+ }
+ else if (resultRelInfo->ri_TrigDesc != NULL &&
+ (resultRelInfo->ri_TrigDesc->trig_insert_after_row ||
+ resultRelInfo->ri_TrigDesc->trig_insert_new_table))
+ {
+ ExecARInsertTriggers(miinfo->estate, resultRelInfo,
+ buffer->slots[i], NIL,
+ mtstate->mt_transition_capture);
+ }
+
+ ExecClearTuple(buffer->slots[i]);
+ }
+ buffer->nused = 0;
+ }
+
+ // TODO: MAX_PARTITION_BUFFERS
+
+ miinfo->bufferedTuples = 0;
+ miinfo->bufferedBytes = 0;
+}
+
+
/* ----------------------------------------------------------------
* ExecInsert
*
@@ -391,6 +451,7 @@ ExecInsert(ModifyTableState *mtstate,
ModifyTable *node = (ModifyTable *) mtstate->ps.plan;
OnConflictAction onconflict = node->onConflictAction;
PartitionTupleRouting *proute = mtstate->mt_partition_tuple_routing;
+ TupleTableSlot *batchslot = NULL;
/*
* If the input result relation is a partitioned table, find the leaf
@@ -631,6 +692,34 @@ ExecInsert(ModifyTableState *mtstate,
/* Since there was no insertion conflict, we're done */
}
+ else if (mtstate->miinfo != NULL &&
+ /* Has unsupported trigger? */
+ (resultRelInfo->ri_TrigDesc == NULL ||
+ (
+ // !resultRelInfo->ri_TrigDesc->trig_insert_before_row
+ !resultRelInfo->ri_TrigDesc->trig_insert_after_row
+ // && !resultRelInfo->ri_TrigDesc->trig_insert_instead_row
+ && !resultRelInfo->ri_TrigDesc->trig_insert_after_statement
+ )
+ ) &&
+ /* Has Fdw? */
+ !resultRelInfo->ri_FdwRoutine
+ // !cstate->volatile_defexprs)
+ // !(contain_volatile_functions(cstate->whereClause))
+ )
+ {
+ if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL)
+ CopyMultiInsertInfoSetupBuffer(mtstate->miinfo,
+ resultRelInfo);
+
+ batchslot = CopyMultiInsertInfoNextFreeSlot(mtstate->miinfo, resultRelInfo);
+ ExecCopySlot(batchslot, slot);
+
+ CopyMultiInsertInfoStore(mtstate->miinfo, resultRelInfo, batchslot, 0, 0); // XXX: tuplen/lineno
+
+ if (CopyMultiInsertInfoIsFull(mtstate->miinfo))
+ InsertMultiInsertInfoFlush(mtstate->miinfo);
+ }
else
{
if (proute && mtstate->prevResultRelInfo != resultRelInfo)
@@ -640,6 +729,13 @@ ExecInsert(ModifyTableState *mtstate,
mtstate->prevResultRelInfo = resultRelInfo;
}
+ /*
+ * Flush pending inserts if this partition can't use
+ * batching, so rows are visible to triggers etc.
+ */
+ if (mtstate->miinfo)
+ InsertMultiInsertInfoFlush(mtstate->miinfo);
+
/* insert the tuple normally */
table_tuple_insert(resultRelationDesc, slot,
estate->es_output_cid,
@@ -659,32 +755,36 @@ ExecInsert(ModifyTableState *mtstate,
setLastTid(&slot->tts_tid);
}
- /*
- * If this insert is the result of a partition key update that moved the
- * tuple to a new partition, put this row into the transition NEW TABLE,
- * if there is one. We need to do this separately for DELETE and INSERT
- * because they happen on different tables.
- */
- ar_insert_trig_tcs = mtstate->mt_transition_capture;
- if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture
- && mtstate->mt_transition_capture->tcs_update_new_table)
+ /* Triggers were already run in the batch insert case */
+ if (batchslot == NULL)
{
- ExecARUpdateTriggers(estate, resultRelInfo, NULL,
- NULL,
- slot,
- NULL,
- mtstate->mt_transition_capture);
-
/*
- * We've already captured the NEW TABLE row, so make sure any AR
- * INSERT trigger fired below doesn't capture it again.
+ * If this insert is the result of a partition key update that moved the
+ * tuple to a new partition, put this row into the transition NEW TABLE,
+ * if there is one. We need to do this separately for DELETE and INSERT
+ * because they happen on different tables.
*/
- ar_insert_trig_tcs = NULL;
- }
+ ar_insert_trig_tcs = mtstate->mt_transition_capture;
+ if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture
+ && mtstate->mt_transition_capture->tcs_update_new_table)
+ {
+ ExecARUpdateTriggers(estate, resultRelInfo, NULL,
+ NULL,
+ slot,
+ NULL,
+ mtstate->mt_transition_capture);
- /* AFTER ROW INSERT Triggers */
- ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes,
- ar_insert_trig_tcs);
+ /*
+ * We've already captured the NEW TABLE row, so make sure any AR
+ * INSERT trigger fired below doesn't capture it again.
+ */
+ ar_insert_trig_tcs = NULL;
+ }
+
+ /* AFTER ROW INSERT Triggers */
+ ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes,
+ ar_insert_trig_tcs);
+ }
list_free(recheckIndexes);
@@ -2242,9 +2342,25 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
mtstate->mt_arowmarks = (List **) palloc0(sizeof(List *) * nplans);
mtstate->mt_nplans = nplans;
mtstate->bistate = NULL;
+ mtstate->miinfo = NULL;
if (operation == CMD_INSERT && insert_in_bulk)
+ {
mtstate->bistate = GetBulkInsertState();
+ /*
+ * For partitioned tables we don't support multi-inserts when there
+ * are any statement level insert triggers.
+ */
+ if (node->rootRelation == 0 ||
+ !mtstate->rootResultRelInfo->ri_TrigDesc ||
+ !mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_new_table)
+ {
+ mtstate->miinfo = calloc(sizeof(*mtstate->miinfo), 1);
+ CopyMultiInsertInfoInit(mtstate->miinfo, mtstate->rootResultRelInfo,
+ (void*)mtstate, estate, GetCurrentCommandId(true), 0);
+ }
+ }
+
/* set up epqstate with dummy subplan data for the moment */
EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, node->epqParam);
mtstate->fireBSTriggers = true;
@@ -2716,6 +2832,13 @@ ExecEndModifyTable(ModifyTableState *node)
table_finish_bulk_insert(node->rootResultRelInfo->ri_RelationDesc, 0);
}
+ if (node->miinfo)
+ {
+ if (!CopyMultiInsertInfoIsEmpty(node->miinfo))
+ InsertMultiInsertInfoFlush(node->miinfo);
+ CopyMultiInsertInfoCleanup(node->miinfo);
+ }
+
/*
* Close all the partitioned tables, leaf partitions, and their indices
* and release the slot used for tuple routing, if set.
diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h
index c639833565..8b03f69cf5 100644
--- a/src/include/commands/copy.h
+++ b/src/include/commands/copy.h
@@ -14,6 +14,7 @@
#ifndef COPY_H
#define COPY_H
+#include "executor/executor.h"
#include "nodes/execnodes.h"
#include "nodes/parsenodes.h"
#include "parser/parse_node.h"
@@ -41,4 +42,188 @@ extern uint64 CopyFrom(CopyState cstate);
extern DestReceiver *CreateCopyDestReceiver(void);
+/*
+ * 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 */
+#define MAX_PARTITION_BUFFERS 32
+
+/* Stores multi-insert data related to a single relation in CopyFrom/INSERT. */
+typedef struct CopyMultiInsertBuffer
+{
+ TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */
+ ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */
+ BulkInsertState bistate; /* BulkInsertState for this rel */
+ 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 */
+ CopyState cstate; /* Copy state for this CopyMultiInsertInfo (not used for insert) */
+ EState *estate; /* Executor state used for COPY/INSERT */
+ CommandId mycid; /* Command Id used for COPY/INSERT */
+ int ti_options; /* table insert options */
+} CopyMultiInsertInfo;
+
+void CopyMultiInsertInfoInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
+ CopyState cstate, EState *estate, CommandId mycid,
+ int ti_options);
+void CopyMultiInsertInfoCleanup(CopyMultiInsertInfo *miinfo);
+
+/*
+ * Allocate memory and initialize a new CopyMultiInsertBuffer for this
+ * ResultRelInfo.
+ */
+static inline 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 = GetBulkInsertState();
+ 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);
+}
+
+/*
+ * 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;
+}
+
+/*
+ * Drop used slots and free member for this buffer.
+ *
+ * The buffer must be flushed before cleanup.
+ */
+static inline void
+CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
+ CopyMultiInsertBuffer *buffer)
+{
+ int i;
+
+ /* Ensure buffer was flushed */
+ Assert(buffer->nused == 0);
+
+ /* Remove back-link to ourself */
+ buffer->resultRelInfo->ri_CopyMultiInsertBuffer = NULL;
+
+ FreeBulkInsertState(buffer->bistate);
+
+ /* 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]);
+
+ table_finish_bulk_insert(buffer->resultRelInfo->ri_RelationDesc,
+ miinfo->ti_options);
+
+ pfree(buffer);
+}
+
+/*
+ * 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 = buffer->nused;
+
+ Assert(buffer != NULL);
+ Assert(nused < MAX_BUFFERED_TUPLES);
+
+ 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;
+}
+
#endif /* COPY_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index e034562877..bcb9986013 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -40,8 +40,8 @@ struct ExprState;
struct ExprContext;
struct RangeTblEntry; /* avoid including parsenodes.h here */
struct ExprEvalStep; /* avoid including execExpr.h everywhere */
-struct CopyMultiInsertBuffer;
-
+// struct CopyMultiInsertBuffer;
+// struct CopyMultiInsertInfo;
/* ----------------
* ExprState node
@@ -499,7 +499,7 @@ typedef struct ResultRelInfo
*/
TupleConversionMap *ri_ChildToRootMap;
- /* for use by copy.c when performing multi-inserts */
+ /* for use by copy.c/modifyTable when performing multi-inserts */
struct CopyMultiInsertBuffer *ri_CopyMultiInsertBuffer;
} ResultRelInfo;
@@ -1177,8 +1177,9 @@ typedef struct ModifyTableState
List **mt_arowmarks; /* per-subplan ExecAuxRowMark lists */
EPQState mt_epqstate; /* for evaluating EvalPlanQual rechecks */
bool fireBSTriggers; /* do we need to fire stmt triggers? */
- BulkInsertState bistate; /* state for bulk insert like INSERT SELECT */
+ BulkInsertState bistate; /* state for bulk insert like INSERT SELECT, when miinfo cannot be used */
ResultRelInfo *prevResultRelInfo; /* last child inserted with bistate */
+ struct CopyMultiInsertInfo *miinfo;
/*
* Slot for storing tuples in the root partitioned table's rowtype during
diff --git a/src/test/regress/expected/insert.out b/src/test/regress/expected/insert.out
index da0dae6240..e0c83d7427 100644
--- a/src/test/regress/expected/insert.out
+++ b/src/test/regress/expected/insert.out
@@ -485,6 +485,22 @@ select count(1) from hash_parted;
(1 row)
commit;
+-- test that index was updated
+vacuum analyze hash_parted;
+explain(costs off)
+select * from hash_parted where a=13;
+ QUERY PLAN
+----------------------------------------------------------
+ Index Only Scan using hpart1_a_idx on hpart1 hash_parted
+ Index Cond: (a = 13)
+(2 rows)
+
+select * from hash_parted where a=13;
+ a
+----
+ 13
+(1 row)
+
-- cleanup
drop table range_parted, list_parted;
drop table hash_parted;
diff --git a/src/test/regress/sql/insert.sql b/src/test/regress/sql/insert.sql
index d3a94f053b..99ec18d9a2 100644
--- a/src/test/regress/sql/insert.sql
+++ b/src/test/regress/sql/insert.sql
@@ -292,6 +292,11 @@ select * from hash_parted;
insert into hash_parted select generate_series(1,9999);
select count(1) from hash_parted;
commit;
+-- test that index was updated
+vacuum analyze hash_parted;
+explain(costs off)
+select * from hash_parted where a=13;
+select * from hash_parted where a=13;
-- cleanup
drop table range_parted, list_parted;
--
2.17.0
--Yylu36WmvOXNoKYn--
^ permalink raw reply [nested|flat] 4+ messages in thread
* [PATCH v6 2/3] Make INSERT SELECT use multi_insert
@ 2020-10-25 03:49 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 4+ messages in thread
From: Justin Pryzby @ 2020-10-25 03:49 UTC (permalink / raw)
TODO: CTAS and matview ?
Renames structures;
Move MultipleInsert functions from copyfrom.c to (tentatively) nodeModifyTable.h;
Move cur_lineno and transition_capture into MultiInsertInfo;
See also: 86b85044e823a304d2a265abc030254d39efe7df
---
src/backend/commands/copyfrom.c | 396 ++---------------------
src/backend/commands/copyfromparse.c | 10 +-
src/backend/executor/execMain.c | 2 +-
src/backend/executor/execPartition.c | 2 +-
src/backend/executor/nodeModifyTable.c | 110 +++++--
src/include/commands/copyfrom_internal.h | 3 +-
src/include/executor/nodeModifyTable.h | 367 +++++++++++++++++++++
src/include/nodes/execnodes.h | 14 +-
src/test/regress/expected/insert.out | 16 +
src/test/regress/sql/insert.sql | 5 +
src/tools/pgindent/typedefs.list | 4 +-
11 files changed, 515 insertions(+), 414 deletions(-)
diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index 1b14e9a6eb..5d65458f3b 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -44,54 +44,6 @@
#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 */
-#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 */
- 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 char *limit_printout_length(const char *str);
@@ -109,7 +61,7 @@ CopyFromErrorCallback(void *arg)
char curlineno_str[32];
snprintf(curlineno_str, sizeof(curlineno_str), UINT64_FORMAT,
- cstate->cur_lineno);
+ cstate->miinfo->cur_lineno);
if (cstate->opts.binary)
{
@@ -204,317 +156,6 @@ limit_printout_length(const char *str)
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 = GetBulkInsertState();
- 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)
-{
- MemoryContext oldcontext;
- int i;
- uint64 save_cur_lineno;
- CopyFromState cstate = miinfo->cstate;
- EState *estate = miinfo->estate;
- CommandId mycid = miinfo->mycid;
- int ti_options = miinfo->ti_options;
- bool line_buf_valid = cstate->line_buf_valid;
- int nused = buffer->nused;
- ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
- TupleTableSlot **slots = buffer->slots;
-
- /*
- * Print error context information correctly, if one of the operations
- * below fail.
- */
- cstate->line_buf_valid = false;
- save_cur_lineno = cstate->cur_lineno;
-
- /*
- * 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, NULL,
- NIL);
- 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]);
- }
-
- /* Mark that all slots are free */
- buffer->nused = 0;
-
- /* reset cur_lineno and line_buf_valid to what they were */
- cstate->line_buf_valid = line_buf_valid;
- cstate->cur_lineno = save_cur_lineno;
-}
-
-/*
- * Drop used slots and free member for this buffer.
- *
- * The buffer must be flushed before cleanup.
- */
-static inline void
-CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
- CopyMultiInsertBuffer *buffer)
-{
- int i;
-
- /* Ensure buffer was flushed */
- Assert(buffer->nused == 0);
-
- /* Remove back-link to ourself */
- buffer->resultRelInfo->ri_CopyMultiInsertBuffer = NULL;
-
- FreeBulkInsertState(buffer->bistate);
-
- /* 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]);
-
- table_finish_bulk_insert(buffer->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' is 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)
-{
- ListCell *lc;
-
- foreach(lc, miinfo->multiInsertBuffers)
- {
- CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) lfirst(lc);
-
- CopyMultiInsertBufferFlush(miinfo, buffer);
- }
-
- 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)
- {
- 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 = buffer->nused;
-
- Assert(buffer != NULL);
- Assert(nused < MAX_BUFFERED_TUPLES);
-
- 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.
*/
@@ -536,7 +177,7 @@ CopyFrom(CopyFromState cstate)
int ti_options = 0; /* start with default options for insert */
BulkInsertState bistate = NULL;
CopyInsertMethod insertMethod;
- CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
+ MultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
@@ -723,7 +364,7 @@ CopyFrom(CopyFromState 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, CopyMultiInsertInfoFlush expects that any before row insert
+ * now, MultiInsertInfoFlush expects that any before row insert
* and statement level insert triggers are on the same relation.
*/
insertMethod = CIM_SINGLE;
@@ -771,7 +412,8 @@ CopyFrom(CopyFromState cstate)
else
insertMethod = CIM_MULTI;
- CopyMultiInsertInfoInit(&multiInsertInfo, resultRelInfo, cstate,
+ MultiInsertInfoInit(&multiInsertInfo, resultRelInfo,
+ cstate->transition_capture,
estate, mycid, ti_options);
}
@@ -809,6 +451,7 @@ CopyFrom(CopyFromState cstate)
errcallback.arg = (void *) cstate;
errcallback.previous = error_context_stack;
error_context_stack = &errcallback;
+ cstate->miinfo = &multiInsertInfo;
for (;;)
{
@@ -834,7 +477,7 @@ CopyFrom(CopyFromState cstate)
Assert(resultRelInfo == target_resultRelInfo);
Assert(insertMethod == CIM_MULTI);
- myslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
+ myslot = MultiInsertInfoNextFreeSlot(&multiInsertInfo,
resultRelInfo);
}
@@ -903,18 +546,18 @@ CopyFrom(CopyFromState cstate)
/* Set the multi-insert buffer to use for this partition. */
if (leafpart_use_multi_insert)
{
- if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL)
- CopyMultiInsertInfoSetupBuffer(&multiInsertInfo,
+ if (resultRelInfo->ri_MultiInsertBuffer == NULL)
+ MultiInsertInfoSetupBuffer(&multiInsertInfo,
resultRelInfo);
}
else if (insertMethod == CIM_MULTI_CONDITIONAL &&
- !CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
+ !MultiInsertInfoIsEmpty(&multiInsertInfo))
{
/*
* Flush pending inserts if this partition can't use
* batching, so rows are visible to triggers etc.
*/
- CopyMultiInsertInfoFlush(&multiInsertInfo, resultRelInfo);
+ MultiInsertInfoFlush(&multiInsertInfo, resultRelInfo);
}
if (bistate != NULL)
@@ -960,7 +603,7 @@ CopyFrom(CopyFromState cstate)
/* no other path available for partitioned table */
Assert(insertMethod == CIM_MULTI_CONDITIONAL);
- batchslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
+ batchslot = MultiInsertInfoNextFreeSlot(&multiInsertInfo,
resultRelInfo);
if (map != NULL)
@@ -1040,17 +683,17 @@ CopyFrom(CopyFromState cstate)
ExecMaterializeSlot(myslot);
/* Add this tuple to the tuple buffer */
- CopyMultiInsertInfoStore(&multiInsertInfo,
+ MultiInsertInfoStore(&multiInsertInfo,
resultRelInfo, myslot,
cstate->line_buf.len,
- cstate->cur_lineno);
+ multiInsertInfo.cur_lineno);
/*
* If enough inserts have queued up, then flush all
* buffers out to their tables.
*/
- if (CopyMultiInsertInfoIsFull(&multiInsertInfo))
- CopyMultiInsertInfoFlush(&multiInsertInfo, resultRelInfo);
+ if (MultiInsertInfoIsFull(&multiInsertInfo))
+ MultiInsertInfoFlush(&multiInsertInfo, resultRelInfo);
}
else
{
@@ -1109,8 +752,8 @@ CopyFrom(CopyFromState cstate)
/* Flush any remaining buffered tuples */
if (insertMethod != CIM_SINGLE)
{
- if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
- CopyMultiInsertInfoFlush(&multiInsertInfo, NULL);
+ if (!MultiInsertInfoIsEmpty(&multiInsertInfo))
+ MultiInsertInfoFlush(&multiInsertInfo, NULL);
}
/* Done, clean up */
@@ -1144,7 +787,7 @@ CopyFrom(CopyFromState cstate)
/* Tear down the multi-insert buffer data */
if (insertMethod != CIM_SINGLE)
- CopyMultiInsertInfoCleanup(&multiInsertInfo);
+ MultiInsertInfoCleanup(&multiInsertInfo);
/* Close all the partitioned tables, leaf partitions, and their indices */
if (proute)
@@ -1323,7 +966,6 @@ BeginCopyFrom(ParseState *pstate,
cstate->reached_eof = false;
cstate->eol_type = EOL_UNKNOWN;
cstate->cur_relname = RelationGetRelationName(cstate->rel);
- cstate->cur_lineno = 0;
cstate->cur_attname = NULL;
cstate->cur_attval = NULL;
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 2eb08a260c..6370be3641 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -456,14 +456,14 @@ NextCopyFromRawFields(CopyFromState cstate, char ***fields, int *nfields)
Assert(!cstate->opts.binary);
/* on input just throw the header line away */
- if (cstate->cur_lineno == 0 && cstate->opts.header_line)
+ if (cstate->miinfo->cur_lineno == 0 && cstate->opts.header_line)
{
- cstate->cur_lineno++;
+ cstate->miinfo->cur_lineno++;
if (CopyReadLine(cstate))
return false; /* done */
}
- cstate->cur_lineno++;
+ cstate->miinfo->cur_lineno++;
/* Actually read the line into memory here */
done = CopyReadLine(cstate);
@@ -605,7 +605,7 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
int16 fld_count;
ListCell *cur;
- cstate->cur_lineno++;
+ cstate->miinfo->cur_lineno++;
if (!CopyGetInt16(cstate, &fld_count))
{
@@ -913,7 +913,7 @@ CopyReadLineText(CopyFromState cstate)
* at all --- is cur_lineno a physical or logical count?)
*/
if (in_quote && c == (cstate->eol_type == EOL_NL ? '\n' : '\r'))
- cstate->cur_lineno++;
+ cstate->miinfo->cur_lineno++;
}
/* Process \r */
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 7179f589f9..855a89b570 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1247,7 +1247,7 @@ InitResultRelInfo(ResultRelInfo *resultRelInfo,
* ExecInitRoutingInfo */
resultRelInfo->ri_PartitionTupleSlot = NULL; /* ditto */
resultRelInfo->ri_ChildToRootMap = NULL;
- resultRelInfo->ri_CopyMultiInsertBuffer = NULL;
+ resultRelInfo->ri_MultiInsertBuffer = NULL;
}
/*
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index 86594bd056..1f8ba785db 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -994,7 +994,7 @@ ExecInitRoutingInfo(ModifyTableState *mtstate,
partRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL)
partRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, partRelInfo);
- partRelInfo->ri_CopyMultiInsertBuffer = NULL;
+ partRelInfo->ri_MultiInsertBuffer = NULL;
/*
* Keep track of it in the PartitionTupleRouting->partitions array.
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 26ff964105..f8cd537b24 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -43,6 +43,7 @@
#include "access/xact.h"
#include "catalog/catalog.h"
#include "commands/trigger.h"
+#include "commands/copy.h"
#include "executor/execPartition.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
@@ -362,6 +363,7 @@ ExecComputeStoredGenerated(ResultRelInfo *resultRelInfo,
MemoryContextSwitchTo(oldContext);
}
+
/* ----------------------------------------------------------------
* ExecInsert
*
@@ -391,6 +393,7 @@ ExecInsert(ModifyTableState *mtstate,
ModifyTable *node = (ModifyTable *) mtstate->ps.plan;
OnConflictAction onconflict = node->onConflictAction;
PartitionTupleRouting *proute = mtstate->mt_partition_tuple_routing;
+ TupleTableSlot *batchslot = NULL;
/*
* If the input result relation is a partitioned table, find the leaf
@@ -631,6 +634,34 @@ ExecInsert(ModifyTableState *mtstate,
/* Since there was no insertion conflict, we're done */
}
+ else if (mtstate->miinfo != NULL &&
+ /* Has unsupported trigger? */
+ (resultRelInfo->ri_TrigDesc == NULL ||
+ (
+ // !resultRelInfo->ri_TrigDesc->trig_insert_before_row
+ !resultRelInfo->ri_TrigDesc->trig_insert_after_row
+ // && !resultRelInfo->ri_TrigDesc->trig_insert_instead_row
+ && !resultRelInfo->ri_TrigDesc->trig_insert_after_statement
+ )
+ ) &&
+ /* Has Fdw? */
+ !resultRelInfo->ri_FdwRoutine
+ // !cstate->volatile_defexprs)
+ // !(contain_volatile_functions(cstate->whereClause))
+ )
+ {
+ if (resultRelInfo->ri_MultiInsertBuffer == NULL)
+ MultiInsertInfoSetupBuffer(mtstate->miinfo,
+ resultRelInfo);
+
+ batchslot = MultiInsertInfoNextFreeSlot(mtstate->miinfo, resultRelInfo);
+ ExecCopySlot(batchslot, slot);
+
+ MultiInsertInfoStore(mtstate->miinfo, resultRelInfo, batchslot, 0, 0); // XXX: tuplen/lineno
+
+ if (MultiInsertInfoIsFull(mtstate->miinfo))
+ MultiInsertInfoFlush(mtstate->miinfo, resultRelInfo);
+ }
else
{
if (proute && mtstate->prevResultRelInfo != resultRelInfo)
@@ -640,6 +671,13 @@ ExecInsert(ModifyTableState *mtstate,
mtstate->prevResultRelInfo = resultRelInfo;
}
+ /*
+ * Flush pending inserts if this partition can't use
+ * batching, so rows are visible to triggers etc.
+ */
+ if (mtstate->miinfo)
+ MultiInsertInfoFlush(mtstate->miinfo, resultRelInfo);
+
/* insert the tuple normally */
table_tuple_insert(resultRelationDesc, slot,
estate->es_output_cid,
@@ -659,32 +697,36 @@ ExecInsert(ModifyTableState *mtstate,
setLastTid(&slot->tts_tid);
}
- /*
- * If this insert is the result of a partition key update that moved the
- * tuple to a new partition, put this row into the transition NEW TABLE,
- * if there is one. We need to do this separately for DELETE and INSERT
- * because they happen on different tables.
- */
- ar_insert_trig_tcs = mtstate->mt_transition_capture;
- if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture
- && mtstate->mt_transition_capture->tcs_update_new_table)
+ /* Triggers were already run in the batch insert case */
+ if (batchslot == NULL)
{
- ExecARUpdateTriggers(estate, resultRelInfo, NULL,
- NULL,
- slot,
- NULL,
- mtstate->mt_transition_capture);
-
/*
- * We've already captured the NEW TABLE row, so make sure any AR
- * INSERT trigger fired below doesn't capture it again.
+ * If this insert is the result of a partition key update that moved the
+ * tuple to a new partition, put this row into the transition NEW TABLE,
+ * if there is one. We need to do this separately for DELETE and INSERT
+ * because they happen on different tables.
*/
- ar_insert_trig_tcs = NULL;
- }
+ ar_insert_trig_tcs = mtstate->mt_transition_capture;
+ if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture
+ && mtstate->mt_transition_capture->tcs_update_new_table)
+ {
+ ExecARUpdateTriggers(estate, resultRelInfo, NULL,
+ NULL,
+ slot,
+ NULL,
+ mtstate->mt_transition_capture);
- /* AFTER ROW INSERT Triggers */
- ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes,
- ar_insert_trig_tcs);
+ /*
+ * We've already captured the NEW TABLE row, so make sure any AR
+ * INSERT trigger fired below doesn't capture it again.
+ */
+ ar_insert_trig_tcs = NULL;
+ }
+
+ /* AFTER ROW INSERT Triggers */
+ ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes,
+ ar_insert_trig_tcs);
+ }
list_free(recheckIndexes);
@@ -2242,9 +2284,26 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
mtstate->mt_arowmarks = (List **) palloc0(sizeof(List *) * nplans);
mtstate->mt_nplans = nplans;
mtstate->bistate = NULL;
+ mtstate->miinfo = NULL;
if (operation == CMD_INSERT && insert_in_bulk)
+ {
mtstate->bistate = GetBulkInsertState();
+ /*
+ * For partitioned tables we don't support multi-inserts when there
+ * are any statement level insert triggers.
+ */
+ if (node->rootRelation == 0 ||
+ !mtstate->rootResultRelInfo->ri_TrigDesc ||
+ !mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_new_table)
+ {
+ mtstate->miinfo = calloc(sizeof(*mtstate->miinfo), 1);
+ MultiInsertInfoInit(mtstate->miinfo, mtstate->rootResultRelInfo,
+ mtstate->mt_transition_capture,
+ estate, GetCurrentCommandId(true), 0);
+ }
+ }
+
/* set up epqstate with dummy subplan data for the moment */
EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, node->epqParam);
mtstate->fireBSTriggers = true;
@@ -2716,6 +2775,13 @@ ExecEndModifyTable(ModifyTableState *node)
table_finish_bulk_insert(node->rootResultRelInfo->ri_RelationDesc, 0);
}
+ if (node->miinfo)
+ {
+ if (!MultiInsertInfoIsEmpty(node->miinfo))
+ MultiInsertInfoFlush(node->miinfo, node->resultRelInfo); // root ?
+ MultiInsertInfoCleanup(node->miinfo);
+ }
+
/*
* Close all the partitioned tables, leaf partitions, and their indices
* and release the slot used for tuple routing, if set.
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index c15ea803c3..c13768a9b2 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
#define COPYFROM_INTERNAL_H
#include "commands/copy.h"
+#include "executor/nodeModifyTable.h"
#include "commands/trigger.h"
/*
@@ -92,7 +93,7 @@ typedef struct CopyFromStateData
/* these are just for error messages, see CopyFromErrorCallback */
const char *cur_relname; /* table name for error messages */
- uint64 cur_lineno; /* line number for error messages */
+ MultiInsertInfo *miinfo; /* Link back to miinfo for error callback */
const char *cur_attname; /* current att for error messages */
const char *cur_attval; /* current att value for error messages */
diff --git a/src/include/executor/nodeModifyTable.h b/src/include/executor/nodeModifyTable.h
index 09c312a052..80d82a9715 100644
--- a/src/include/executor/nodeModifyTable.h
+++ b/src/include/executor/nodeModifyTable.h
@@ -13,6 +13,8 @@
#ifndef NODEMODIFYTABLE_H
#define NODEMODIFYTABLE_H
+#include "commands/trigger.h"
+#include "executor/executor.h" // XXX
#include "nodes/execnodes.h"
extern PGDLLIMPORT bool insert_in_bulk;
@@ -25,4 +27,369 @@ extern ModifyTableState *ExecInitModifyTable(ModifyTable *node, EState *estate,
extern void ExecEndModifyTable(ModifyTableState *node);
extern void ExecReScanModifyTable(ModifyTableState *node);
+/* Bulk insert stuff which used to live in copy.c */
+
+/*
+ * No more than this many tuples per MultiInsertBuffer
+ *
+ * Caution: Don't make this too big, as we could end up with this many
+ * MultiInsertBuffer items stored in MultiInsertInfo'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 */
+#define MAX_PARTITION_BUFFERS 32
+
+/* Stores multi-insert data related to a single relation in CopyFrom. */
+typedef struct MultiInsertBuffer
+{
+ TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */
+ ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */
+ BulkInsertState bistate; /* BulkInsertState for this rel */
+ int nused; /* number of 'slots' containing tuples */
+ uint64 linenos[MAX_BUFFERED_TUPLES]; /* Line # of tuple in copy
+ * stream */
+} MultiInsertBuffer;
+
+/*
+ * Stores one or many MultiInsertBuffers 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 MultiInsertInfo
+{
+ List *multiInsertBuffers; /* List of tracked MultiInsertBuffers */
+ int bufferedTuples; /* number of tuples buffered over all buffers */
+ int bufferedBytes; /* number of bytes from all buffered tuples */
+ TransitionCaptureState *transition_capture;
+ EState *estate; /* Executor state used for COPY */
+ CommandId mycid; /* Command Id used for COPY */
+ int ti_options; /* table insert options */
+
+ /* Line number for errors in copyfrom.c */
+ uint64 cur_lineno;
+ bool line_buf_valid;
+} MultiInsertInfo;
+
+
+/*
+ * Allocate memory and initialize a new MultiInsertBuffer for this
+ * ResultRelInfo.
+ */
+static MultiInsertBuffer *
+MultiInsertBufferInit(ResultRelInfo *rri)
+{
+ MultiInsertBuffer *buffer;
+
+ buffer = (MultiInsertBuffer *) palloc(sizeof(MultiInsertBuffer));
+ memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES);
+ buffer->resultRelInfo = rri;
+ buffer->bistate = GetBulkInsertState();
+ buffer->nused = 0;
+
+ return buffer;
+}
+
+/*
+ * Make a new buffer for this ResultRelInfo.
+ */
+static inline void
+MultiInsertInfoSetupBuffer(MultiInsertInfo *miinfo,
+ ResultRelInfo *rri)
+{
+ MultiInsertBuffer *buffer;
+
+ buffer = MultiInsertBufferInit(rri);
+
+ /* Setup back-link so we can easily find this buffer again */
+ rri->ri_MultiInsertBuffer = buffer;
+ /* Record that we're tracking this buffer */
+ miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
+}
+
+/*
+ * Initialize an already allocated MultiInsertInfo.
+ *
+ * If rri is a non-partitioned table then a MultiInsertBuffer is set up
+ * for that table.
+ */
+static inline void
+MultiInsertInfoInit(MultiInsertInfo *miinfo, ResultRelInfo *rri,
+ TransitionCaptureState *transition_capture,
+ EState *estate, CommandId mycid, int ti_options)
+{
+ miinfo->multiInsertBuffers = NIL;
+ miinfo->bufferedTuples = 0;
+ miinfo->bufferedBytes = 0;
+ miinfo->transition_capture = transition_capture;
+ miinfo->estate = estate;
+ miinfo->mycid = mycid;
+ miinfo->ti_options = ti_options;
+ miinfo->cur_lineno = 0;
+
+ /*
+ * 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)
+ MultiInsertInfoSetupBuffer(miinfo, rri);
+}
+
+/*
+ * Returns true if the buffers are full
+ */
+static inline bool
+MultiInsertInfoIsFull(MultiInsertInfo *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
+MultiInsertInfoIsEmpty(MultiInsertInfo *miinfo)
+{
+ return miinfo->bufferedTuples == 0;
+}
+
+/*
+ * Write the tuples stored in 'buffer' out to the table.
+ */
+static inline void
+MultiInsertBufferFlush(MultiInsertInfo *miinfo,
+ MultiInsertBuffer *buffer)
+{
+ MemoryContext oldcontext;
+ int i;
+ uint64 save_cur_lineno;
+ EState *estate = miinfo->estate;
+ CommandId mycid = miinfo->mycid;
+ int ti_options = miinfo->ti_options;
+ bool line_buf_valid = miinfo->line_buf_valid;
+ int nused = buffer->nused;
+ ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
+ TupleTableSlot **slots = buffer->slots;
+
+ /*
+ * Print error context information correctly, if one of the operations
+ * below fail.
+ */
+ miinfo->line_buf_valid = false;
+ save_cur_lineno = miinfo->cur_lineno;
+
+ /*
+ * table_multi_insert may leak memory, so switch to short-lived memory
+ * context before calling it.
+ */
+ oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); // XXX requires executor.h
+ 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;
+
+ miinfo->cur_lineno = buffer->linenos[i];
+ recheckIndexes =
+ ExecInsertIndexTuples(resultRelInfo,
+ buffer->slots[i], estate, false, NULL,
+ NIL);
+ ExecARInsertTriggers(estate, resultRelInfo,
+ slots[i], recheckIndexes,
+ miinfo->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))
+ {
+ miinfo->cur_lineno = buffer->linenos[i];
+ ExecARInsertTriggers(estate, resultRelInfo,
+ slots[i], NIL, miinfo->transition_capture);
+ }
+
+ ExecClearTuple(slots[i]);
+ }
+
+ /* Mark that all slots are free */
+ buffer->nused = 0;
+
+ /* reset cur_lineno and line_buf_valid to what they were */
+ miinfo->line_buf_valid = line_buf_valid;
+ miinfo->cur_lineno = save_cur_lineno;
+}
+
+/*
+ * Drop used slots and free member for this buffer.
+ *
+ * The buffer must be flushed before cleanup.
+ */
+static inline void
+MultiInsertBufferCleanup(MultiInsertInfo *miinfo,
+ MultiInsertBuffer *buffer)
+{
+ int i;
+
+ /* Ensure buffer was flushed */
+ Assert(buffer->nused == 0);
+
+ /* Remove back-link to ourself */
+ buffer->resultRelInfo->ri_MultiInsertBuffer = NULL;
+
+ FreeBulkInsertState(buffer->bistate);
+
+ /* 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]);
+
+ table_finish_bulk_insert(buffer->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' is the ResultRelInfo that's currently being
+ * used. When cleaning up old buffers we'll never remove the one for
+ * 'curr_rri'.
+ */
+static inline void
+MultiInsertInfoFlush(MultiInsertInfo *miinfo, ResultRelInfo *curr_rri)
+{
+ ListCell *lc;
+
+ foreach(lc, miinfo->multiInsertBuffers)
+ {
+ MultiInsertBuffer *buffer = (MultiInsertBuffer *) lfirst(lc);
+
+ MultiInsertBufferFlush(miinfo, buffer);
+ }
+
+ 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)
+ {
+ MultiInsertBuffer *buffer;
+
+ buffer = (MultiInsertBuffer *) 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)
+ {
+ miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
+ miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
+ buffer = (MultiInsertBuffer *) linitial(miinfo->multiInsertBuffers);
+ }
+
+ MultiInsertBufferCleanup(miinfo, buffer);
+ miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
+ }
+}
+
+/*
+ * Cleanup allocated buffers and free memory
+ */
+static inline void
+MultiInsertInfoCleanup(MultiInsertInfo *miinfo)
+{
+ ListCell *lc;
+
+ foreach(lc, miinfo->multiInsertBuffers)
+ MultiInsertBufferCleanup(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 *
+MultiInsertInfoNextFreeSlot(MultiInsertInfo *miinfo,
+ ResultRelInfo *rri)
+{
+ MultiInsertBuffer *buffer = rri->ri_MultiInsertBuffer;
+ int nused = buffer->nused;
+
+ Assert(buffer != NULL);
+ Assert(nused < MAX_BUFFERED_TUPLES);
+
+ 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
+ * MultiInsertInfoNextFreeSlot as being consumed.
+ */
+static inline void
+MultiInsertInfoStore(MultiInsertInfo *miinfo, ResultRelInfo *rri,
+ TupleTableSlot *slot, int tuplen, uint64 lineno)
+{
+ MultiInsertBuffer *buffer = rri->ri_MultiInsertBuffer;
+
+ 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;
+}
+
#endif /* NODEMODIFYTABLE_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index c79d13aa44..985a98f854 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -33,6 +33,9 @@
#include "utils/tuplesort.h"
#include "utils/tuplestore.h"
+/* This would be a circular inclusion */
+// #include "executor/nodeModifyTable.h"
+
struct PlanState; /* forward references in this file */
struct ParallelHashJoinState;
struct ExecRowMark;
@@ -40,8 +43,8 @@ struct ExprState;
struct ExprContext;
struct RangeTblEntry; /* avoid including parsenodes.h here */
struct ExprEvalStep; /* avoid including execExpr.h everywhere */
-struct CopyMultiInsertBuffer;
-
+// struct MultiInsertBuffer;
+// struct MultiInsertInfo;
/* ----------------
* ExprState node
@@ -499,8 +502,8 @@ typedef struct ResultRelInfo
*/
TupleConversionMap *ri_ChildToRootMap;
- /* for use by copyfrom.c when performing multi-inserts */
- struct CopyMultiInsertBuffer *ri_CopyMultiInsertBuffer;
+ /* for use by copyfrom.c/modifyTable when performing multi-inserts */
+ struct MultiInsertBuffer *ri_MultiInsertBuffer;
} ResultRelInfo;
/* ----------------
@@ -1177,8 +1180,9 @@ typedef struct ModifyTableState
List **mt_arowmarks; /* per-subplan ExecAuxRowMark lists */
EPQState mt_epqstate; /* for evaluating EvalPlanQual rechecks */
bool fireBSTriggers; /* do we need to fire stmt triggers? */
- BulkInsertState bistate; /* state for bulk insert like INSERT SELECT */
+ BulkInsertState bistate; /* state for bulk insert like INSERT SELECT, when miinfo cannot be used */
ResultRelInfo *prevResultRelInfo; /* last child inserted with bistate */
+ struct MultiInsertInfo *miinfo;
/*
* Slot for storing tuples in the root partitioned table's rowtype during
diff --git a/src/test/regress/expected/insert.out b/src/test/regress/expected/insert.out
index da0dae6240..e0c83d7427 100644
--- a/src/test/regress/expected/insert.out
+++ b/src/test/regress/expected/insert.out
@@ -485,6 +485,22 @@ select count(1) from hash_parted;
(1 row)
commit;
+-- test that index was updated
+vacuum analyze hash_parted;
+explain(costs off)
+select * from hash_parted where a=13;
+ QUERY PLAN
+----------------------------------------------------------
+ Index Only Scan using hpart1_a_idx on hpart1 hash_parted
+ Index Cond: (a = 13)
+(2 rows)
+
+select * from hash_parted where a=13;
+ a
+----
+ 13
+(1 row)
+
-- cleanup
drop table range_parted, list_parted;
drop table hash_parted;
diff --git a/src/test/regress/sql/insert.sql b/src/test/regress/sql/insert.sql
index d3a94f053b..99ec18d9a2 100644
--- a/src/test/regress/sql/insert.sql
+++ b/src/test/regress/sql/insert.sql
@@ -292,6 +292,11 @@ select * from hash_parted;
insert into hash_parted select generate_series(1,9999);
select count(1) from hash_parted;
commit;
+-- test that index was updated
+vacuum analyze hash_parted;
+explain(costs off)
+select * from hash_parted where a=13;
+select * from hash_parted where a=13;
-- cleanup
drop table range_parted, list_parted;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index fde701bfd4..1bc9db8af4 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -423,8 +423,6 @@ ConvertRowtypeExpr
CookedConstraint
CopyDest
CopyInsertMethod
-CopyMultiInsertBuffer
-CopyMultiInsertInfo
CopyState
CopyStateData
CopyStmt
@@ -1389,6 +1387,8 @@ ModifyTableState
MorphOpaque
MsgType
MultiAssignRef
+MultiInsertBuffer
+MultiInsertInfo
MultiSortSupport
MultiSortSupportData
MultiXactId
--
2.17.0
--V0207lvV8h4k8FAm
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v6-0003-Dynamically-switch-to-multi-insert-mode.patch"
^ permalink raw reply [nested|flat] 4+ messages in thread
* [PATCH v7 2/3] Make INSERT SELECT use multi_insert
@ 2020-10-25 03:49 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 4+ messages in thread
From: Justin Pryzby @ 2020-10-25 03:49 UTC (permalink / raw)
Renames structures;
Move MultipleInsert functions from copyfrom.c to (tentatively) nodeModifyTable.h;
Move into MultiInsertInfo: transition_capture and cur_lineno (via cstate->miinfo);
See also: 86b85044e823a304d2a265abc030254d39efe7df
---
src/backend/commands/copyfrom.c | 394 +----------------------
src/backend/commands/copyfromparse.c | 10 +-
src/backend/executor/execMain.c | 2 +-
src/backend/executor/execPartition.c | 2 +-
src/backend/executor/nodeModifyTable.c | 156 +++++++--
src/include/commands/copyfrom_internal.h | 5 +-
src/include/executor/nodeModifyTable.h | 367 +++++++++++++++++++++
src/include/nodes/execnodes.h | 14 +-
src/test/regress/expected/insert.out | 16 +
src/test/regress/sql/insert.sql | 5 +
src/tools/pgindent/typedefs.list | 4 +-
11 files changed, 559 insertions(+), 416 deletions(-)
diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index 1b14e9a6eb..c4fe75df8e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -44,54 +44,6 @@
#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 */
-#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 */
- 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 char *limit_printout_length(const char *str);
@@ -109,7 +61,7 @@ CopyFromErrorCallback(void *arg)
char curlineno_str[32];
snprintf(curlineno_str, sizeof(curlineno_str), UINT64_FORMAT,
- cstate->cur_lineno);
+ cstate->miinfo.cur_lineno);
if (cstate->opts.binary)
{
@@ -204,317 +156,6 @@ limit_printout_length(const char *str)
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 = GetBulkInsertState();
- 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)
-{
- MemoryContext oldcontext;
- int i;
- uint64 save_cur_lineno;
- CopyFromState cstate = miinfo->cstate;
- EState *estate = miinfo->estate;
- CommandId mycid = miinfo->mycid;
- int ti_options = miinfo->ti_options;
- bool line_buf_valid = cstate->line_buf_valid;
- int nused = buffer->nused;
- ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
- TupleTableSlot **slots = buffer->slots;
-
- /*
- * Print error context information correctly, if one of the operations
- * below fail.
- */
- cstate->line_buf_valid = false;
- save_cur_lineno = cstate->cur_lineno;
-
- /*
- * 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, NULL,
- NIL);
- 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]);
- }
-
- /* Mark that all slots are free */
- buffer->nused = 0;
-
- /* reset cur_lineno and line_buf_valid to what they were */
- cstate->line_buf_valid = line_buf_valid;
- cstate->cur_lineno = save_cur_lineno;
-}
-
-/*
- * Drop used slots and free member for this buffer.
- *
- * The buffer must be flushed before cleanup.
- */
-static inline void
-CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo,
- CopyMultiInsertBuffer *buffer)
-{
- int i;
-
- /* Ensure buffer was flushed */
- Assert(buffer->nused == 0);
-
- /* Remove back-link to ourself */
- buffer->resultRelInfo->ri_CopyMultiInsertBuffer = NULL;
-
- FreeBulkInsertState(buffer->bistate);
-
- /* 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]);
-
- table_finish_bulk_insert(buffer->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' is 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)
-{
- ListCell *lc;
-
- foreach(lc, miinfo->multiInsertBuffers)
- {
- CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) lfirst(lc);
-
- CopyMultiInsertBufferFlush(miinfo, buffer);
- }
-
- 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)
- {
- 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 = buffer->nused;
-
- Assert(buffer != NULL);
- Assert(nused < MAX_BUFFERED_TUPLES);
-
- 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.
*/
@@ -536,7 +177,6 @@ CopyFrom(CopyFromState cstate)
int ti_options = 0; /* start with default options for insert */
BulkInsertState bistate = NULL;
CopyInsertMethod insertMethod;
- CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */
uint64 processed = 0;
bool has_before_insert_row_trig;
bool has_instead_insert_row_trig;
@@ -723,7 +363,7 @@ CopyFrom(CopyFromState 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, CopyMultiInsertInfoFlush expects that any before row insert
+ * now, MultiInsertInfoFlush expects that any before row insert
* and statement level insert triggers are on the same relation.
*/
insertMethod = CIM_SINGLE;
@@ -771,7 +411,8 @@ CopyFrom(CopyFromState cstate)
else
insertMethod = CIM_MULTI;
- CopyMultiInsertInfoInit(&multiInsertInfo, resultRelInfo, cstate,
+ MultiInsertInfoInit(&cstate->miinfo, resultRelInfo,
+ cstate->transition_capture,
estate, mycid, ti_options);
}
@@ -834,7 +475,7 @@ CopyFrom(CopyFromState cstate)
Assert(resultRelInfo == target_resultRelInfo);
Assert(insertMethod == CIM_MULTI);
- myslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
+ myslot = MultiInsertInfoNextFreeSlot(&cstate->miinfo,
resultRelInfo);
}
@@ -903,18 +544,18 @@ CopyFrom(CopyFromState cstate)
/* Set the multi-insert buffer to use for this partition. */
if (leafpart_use_multi_insert)
{
- if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL)
- CopyMultiInsertInfoSetupBuffer(&multiInsertInfo,
+ if (resultRelInfo->ri_MultiInsertBuffer == NULL)
+ MultiInsertInfoSetupBuffer(&cstate->miinfo,
resultRelInfo);
}
else if (insertMethod == CIM_MULTI_CONDITIONAL &&
- !CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
+ !MultiInsertInfoIsEmpty(&cstate->miinfo))
{
/*
* Flush pending inserts if this partition can't use
* batching, so rows are visible to triggers etc.
*/
- CopyMultiInsertInfoFlush(&multiInsertInfo, resultRelInfo);
+ MultiInsertInfoFlush(&cstate->miinfo, resultRelInfo);
}
if (bistate != NULL)
@@ -960,7 +601,7 @@ CopyFrom(CopyFromState cstate)
/* no other path available for partitioned table */
Assert(insertMethod == CIM_MULTI_CONDITIONAL);
- batchslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo,
+ batchslot = MultiInsertInfoNextFreeSlot(&cstate->miinfo,
resultRelInfo);
if (map != NULL)
@@ -1040,17 +681,17 @@ CopyFrom(CopyFromState cstate)
ExecMaterializeSlot(myslot);
/* Add this tuple to the tuple buffer */
- CopyMultiInsertInfoStore(&multiInsertInfo,
+ MultiInsertInfoStore(&cstate->miinfo,
resultRelInfo, myslot,
cstate->line_buf.len,
- cstate->cur_lineno);
+ cstate->miinfo.cur_lineno);
/*
* If enough inserts have queued up, then flush all
* buffers out to their tables.
*/
- if (CopyMultiInsertInfoIsFull(&multiInsertInfo))
- CopyMultiInsertInfoFlush(&multiInsertInfo, resultRelInfo);
+ if (MultiInsertInfoIsFull(&cstate->miinfo))
+ MultiInsertInfoFlush(&cstate->miinfo, resultRelInfo);
}
else
{
@@ -1109,8 +750,8 @@ CopyFrom(CopyFromState cstate)
/* Flush any remaining buffered tuples */
if (insertMethod != CIM_SINGLE)
{
- if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo))
- CopyMultiInsertInfoFlush(&multiInsertInfo, NULL);
+ if (!MultiInsertInfoIsEmpty(&cstate->miinfo))
+ MultiInsertInfoFlush(&cstate->miinfo, NULL);
}
/* Done, clean up */
@@ -1144,7 +785,7 @@ CopyFrom(CopyFromState cstate)
/* Tear down the multi-insert buffer data */
if (insertMethod != CIM_SINGLE)
- CopyMultiInsertInfoCleanup(&multiInsertInfo);
+ MultiInsertInfoCleanup(&cstate->miinfo);
/* Close all the partitioned tables, leaf partitions, and their indices */
if (proute)
@@ -1323,7 +964,6 @@ BeginCopyFrom(ParseState *pstate,
cstate->reached_eof = false;
cstate->eol_type = EOL_UNKNOWN;
cstate->cur_relname = RelationGetRelationName(cstate->rel);
- cstate->cur_lineno = 0;
cstate->cur_attname = NULL;
cstate->cur_attval = NULL;
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index b7a37bcdbd..6eb3c1be79 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -456,14 +456,14 @@ NextCopyFromRawFields(CopyFromState cstate, char ***fields, int *nfields)
Assert(!cstate->opts.binary);
/* on input just throw the header line away */
- if (cstate->cur_lineno == 0 && cstate->opts.header_line)
+ if (cstate->miinfo.cur_lineno == 0 && cstate->opts.header_line)
{
- cstate->cur_lineno++;
+ cstate->miinfo.cur_lineno++;
if (CopyReadLine(cstate))
return false; /* done */
}
- cstate->cur_lineno++;
+ cstate->miinfo.cur_lineno++;
/* Actually read the line into memory here */
done = CopyReadLine(cstate);
@@ -605,7 +605,7 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
int16 fld_count;
ListCell *cur;
- cstate->cur_lineno++;
+ cstate->miinfo.cur_lineno++;
if (!CopyGetInt16(cstate, &fld_count))
{
@@ -913,7 +913,7 @@ CopyReadLineText(CopyFromState cstate)
* at all --- is cur_lineno a physical or logical count?)
*/
if (in_quote && c == (cstate->eol_type == EOL_NL ? '\n' : '\r'))
- cstate->cur_lineno++;
+ cstate->miinfo.cur_lineno++;
}
/* Process \r */
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 7179f589f9..855a89b570 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1247,7 +1247,7 @@ InitResultRelInfo(ResultRelInfo *resultRelInfo,
* ExecInitRoutingInfo */
resultRelInfo->ri_PartitionTupleSlot = NULL; /* ditto */
resultRelInfo->ri_ChildToRootMap = NULL;
- resultRelInfo->ri_CopyMultiInsertBuffer = NULL;
+ resultRelInfo->ri_MultiInsertBuffer = NULL;
}
/*
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index 86594bd056..1f8ba785db 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -994,7 +994,7 @@ ExecInitRoutingInfo(ModifyTableState *mtstate,
partRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL)
partRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, partRelInfo);
- partRelInfo->ri_CopyMultiInsertBuffer = NULL;
+ partRelInfo->ri_MultiInsertBuffer = NULL;
/*
* Keep track of it in the PartitionTupleRouting->partitions array.
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index f65ae2c0d6..05f70f140e 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -43,6 +43,7 @@
#include "access/xact.h"
#include "catalog/catalog.h"
#include "commands/trigger.h"
+#include "commands/copy.h"
#include "executor/execPartition.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
@@ -391,6 +392,8 @@ ExecInsert(ModifyTableState *mtstate,
ModifyTable *node = (ModifyTable *) mtstate->ps.plan;
OnConflictAction onconflict = node->onConflictAction;
PartitionTupleRouting *proute = mtstate->mt_partition_tuple_routing;
+ TupleTableSlot *batchslot = NULL;
+ bool use_multi_insert = false;
/*
* If the input result relation is a partitioned table, find the leaf
@@ -410,6 +413,37 @@ ExecInsert(ModifyTableState *mtstate,
resultRelationDesc = resultRelInfo->ri_RelationDesc;
+ if (!mtstate->miinfo ||
+ mtstate->operation != CMD_INSERT || onconflict != ONCONFLICT_NONE)
+ ; /* If multi-inserts aren't possible at all, don't check further .. */
+ else if (proute == NULL)
+ use_multi_insert = true;
+ else
+ {
+ /*
+ * If a partitioned table itself allows multi-insert, and bistate
+ * indicates we've inserted the threshold number of tuples, check if
+ * the partition also supports it.
+ */
+
+ /* Determine which triggers exist on this partition */
+ // XXX copyfrom.c only checks triggers when the partition changes,
+ // so maybe use_multi_insert should be in mtstate ?
+ bool has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc &&
+ resultRelInfo->ri_TrigDesc->trig_insert_before_row);
+
+ bool 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 partition.
+ */
+ use_multi_insert = !has_before_insert_row_trig &&
+ !has_instead_insert_row_trig &&
+ resultRelInfo->ri_FdwRoutine == NULL;
+ }
+
/*
* BEFORE ROW INSERT Triggers.
*
@@ -631,6 +665,19 @@ ExecInsert(ModifyTableState *mtstate,
/* Since there was no insertion conflict, we're done */
}
+ else if (use_multi_insert)
+ {
+ if (resultRelInfo->ri_MultiInsertBuffer == NULL)
+ MultiInsertInfoSetupBuffer(mtstate->miinfo, resultRelInfo);
+
+ batchslot = MultiInsertInfoNextFreeSlot(mtstate->miinfo, resultRelInfo);
+ ExecCopySlot(batchslot, slot);
+
+ MultiInsertInfoStore(mtstate->miinfo, resultRelInfo, batchslot, 0, 0); // XXX: tuplen/lineno
+
+ if (MultiInsertInfoIsFull(mtstate->miinfo))
+ MultiInsertInfoFlush(mtstate->miinfo, resultRelInfo);
+ }
else
{
if (proute && mtstate->prevResultRelInfo != resultRelInfo)
@@ -640,6 +687,13 @@ ExecInsert(ModifyTableState *mtstate,
mtstate->prevResultRelInfo = resultRelInfo;
}
+ /*
+ * Flush pending inserts if this partition can't use
+ * batching, so rows are visible to triggers etc.
+ */
+ if (mtstate->miinfo)
+ MultiInsertInfoFlush(mtstate->miinfo, resultRelInfo);
+
/* insert the tuple normally */
table_tuple_insert(resultRelationDesc, slot,
estate->es_output_cid,
@@ -656,32 +710,36 @@ ExecInsert(ModifyTableState *mtstate,
if (canSetTag)
(estate->es_processed)++;
- /*
- * If this insert is the result of a partition key update that moved the
- * tuple to a new partition, put this row into the transition NEW TABLE,
- * if there is one. We need to do this separately for DELETE and INSERT
- * because they happen on different tables.
- */
- ar_insert_trig_tcs = mtstate->mt_transition_capture;
- if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture
- && mtstate->mt_transition_capture->tcs_update_new_table)
+ /* Triggers were already run in the batch insert case */
+ if (batchslot == NULL)
{
- ExecARUpdateTriggers(estate, resultRelInfo, NULL,
- NULL,
- slot,
- NULL,
- mtstate->mt_transition_capture);
-
/*
- * We've already captured the NEW TABLE row, so make sure any AR
- * INSERT trigger fired below doesn't capture it again.
+ * If this insert is the result of a partition key update that moved the
+ * tuple to a new partition, put this row into the transition NEW TABLE,
+ * if there is one. We need to do this separately for DELETE and INSERT
+ * because they happen on different tables.
*/
- ar_insert_trig_tcs = NULL;
- }
+ ar_insert_trig_tcs = mtstate->mt_transition_capture;
+ if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture
+ && mtstate->mt_transition_capture->tcs_update_new_table)
+ {
+ ExecARUpdateTriggers(estate, resultRelInfo, NULL,
+ NULL,
+ slot,
+ NULL,
+ mtstate->mt_transition_capture);
- /* AFTER ROW INSERT Triggers */
- ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes,
- ar_insert_trig_tcs);
+ /*
+ * We've already captured the NEW TABLE row, so make sure any AR
+ * INSERT trigger fired below doesn't capture it again.
+ */
+ ar_insert_trig_tcs = NULL;
+ }
+
+ /* AFTER ROW INSERT Triggers */
+ ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes,
+ ar_insert_trig_tcs);
+ }
list_free(recheckIndexes);
@@ -2238,9 +2296,52 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
mtstate->mt_arowmarks = (List **) palloc0(sizeof(List *) * nplans);
mtstate->mt_nplans = nplans;
- mtstate->bistate = NULL;
- if (operation == CMD_INSERT && insert_in_bulk)
+
+ if (insert_in_bulk && operation == CMD_INSERT &&
+ node->onConflictAction == ONCONFLICT_NONE)
mtstate->bistate = GetBulkInsertState();
+ else
+ mtstate->bistate = NULL;
+
+ /*
+ * Set miinfo if it can support multi-insert. This is the equivalent of
+ * CIM_SINGLE et al in copyfrom.c
+ */
+
+ if (operation != CMD_INSERT ||
+ node->onConflictAction != ONCONFLICT_NONE ||
+ !insert_in_bulk)
+ mtstate->miinfo = NULL;
+ else if (mtstate->rootResultRelInfo->ri_TrigDesc != NULL &&
+ (mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_before_row ||
+ // mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_after_row || // XXX or any row level triggers at all?
+ mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_instead_row))
+ /*
+ * Can't support multi-inserts when there are any BEFORE/INSTEAD OF
+ * triggers on the table.
+ */
+ mtstate->miinfo = NULL;
+ else if (node->rootRelation > 0 &&
+ mtstate->rootResultRelInfo->ri_TrigDesc != NULL &&
+ mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_new_table)
+
+ /*
+ * For partitioned tables we can't support multi-inserts when there
+ * are any statement level insert triggers.
+ */
+ mtstate->miinfo = NULL;
+ else if (mtstate->rootResultRelInfo->ri_FdwRoutine != NULL
+ /* || cstate->volatile_defexprs */ )
+ // XXX contain_volatile_functions_not_nextval((Node *) defexpr);
+ /* Can't support multi-inserts to foreign tables or if there are any */
+ mtstate->miinfo = NULL;
+ else
+ {
+ mtstate->miinfo = calloc(sizeof(*mtstate->miinfo), 1);
+ MultiInsertInfoInit(mtstate->miinfo, mtstate->rootResultRelInfo,
+ mtstate->mt_transition_capture,
+ estate, GetCurrentCommandId(true), 0);
+ }
/* set up epqstate with dummy subplan data for the moment */
EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, node->epqParam);
@@ -2713,6 +2814,13 @@ ExecEndModifyTable(ModifyTableState *node)
table_finish_bulk_insert(node->rootResultRelInfo->ri_RelationDesc, 0);
}
+ if (node->miinfo)
+ {
+ if (!MultiInsertInfoIsEmpty(node->miinfo))
+ MultiInsertInfoFlush(node->miinfo, node->resultRelInfo); // root ?
+ MultiInsertInfoCleanup(node->miinfo);
+ }
+
/*
* Close all the partitioned tables, leaf partitions, and their indices
* and release the slot used for tuple routing, if set.
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index c15ea803c3..c0603e13ea 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
#define COPYFROM_INTERNAL_H
#include "commands/copy.h"
+#include "executor/nodeModifyTable.h"
#include "commands/trigger.h"
/*
@@ -92,10 +93,12 @@ typedef struct CopyFromStateData
/* these are just for error messages, see CopyFromErrorCallback */
const char *cur_relname; /* table name for error messages */
- uint64 cur_lineno; /* line number for error messages */
const char *cur_attname; /* current att for error messages */
const char *cur_attval; /* current att value for error messages */
+ /* For bulk inserts and for error callback */
+ MultiInsertInfo miinfo;
+
/*
* Working state
*/
diff --git a/src/include/executor/nodeModifyTable.h b/src/include/executor/nodeModifyTable.h
index 09c312a052..ebe62c2e40 100644
--- a/src/include/executor/nodeModifyTable.h
+++ b/src/include/executor/nodeModifyTable.h
@@ -13,6 +13,8 @@
#ifndef NODEMODIFYTABLE_H
#define NODEMODIFYTABLE_H
+#include "commands/trigger.h"
+#include "executor/executor.h" // XXX
#include "nodes/execnodes.h"
extern PGDLLIMPORT bool insert_in_bulk;
@@ -25,4 +27,369 @@ extern ModifyTableState *ExecInitModifyTable(ModifyTable *node, EState *estate,
extern void ExecEndModifyTable(ModifyTableState *node);
extern void ExecReScanModifyTable(ModifyTableState *node);
+/* Bulk insert stuff which used to live in copy.c */
+
+/*
+ * No more than this many tuples per MultiInsertBuffer
+ *
+ * Caution: Don't make this too big, as we could end up with this many
+ * MultiInsertBuffer items stored in MultiInsertInfo'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 */
+#define MAX_PARTITION_BUFFERS 32
+
+/* Stores multi-insert data related to a single relation in CopyFrom. */
+typedef struct MultiInsertBuffer
+{
+ TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */
+ ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */
+ BulkInsertState bistate; /* BulkInsertState for this rel */
+ int nused; /* number of 'slots' containing tuples */
+ uint64 linenos[MAX_BUFFERED_TUPLES]; /* Line # of tuple in copy
+ * stream */
+} MultiInsertBuffer;
+
+/*
+ * Stores one or many MultiInsertBuffers and details about the size and
+ * number of tuples which are stored in them. This allows multiple buffers to
+ * exist at once when COPY/INSERTing into a partitioned table.
+ */
+typedef struct MultiInsertInfo
+{
+ List *multiInsertBuffers; /* List of tracked MultiInsertBuffers */
+ int bufferedTuples; /* number of tuples buffered over all buffers */
+ int bufferedBytes; /* number of bytes from all buffered tuples */
+ TransitionCaptureState *transition_capture;
+ EState *estate; /* Executor state */
+ CommandId mycid; /* Command Id */
+ int ti_options; /* table insert options */
+
+ /* Line number for errors in copyfrom.c */
+ uint64 cur_lineno;
+ bool line_buf_valid;
+} MultiInsertInfo;
+
+
+/*
+ * Allocate memory and initialize a new MultiInsertBuffer for this
+ * ResultRelInfo.
+ */
+static MultiInsertBuffer *
+MultiInsertBufferInit(ResultRelInfo *rri)
+{
+ MultiInsertBuffer *buffer;
+
+ buffer = (MultiInsertBuffer *) palloc(sizeof(MultiInsertBuffer));
+ memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES);
+ buffer->resultRelInfo = rri;
+ buffer->bistate = GetBulkInsertState();
+ buffer->nused = 0;
+
+ return buffer;
+}
+
+/*
+ * Make a new buffer for this ResultRelInfo.
+ */
+static inline void
+MultiInsertInfoSetupBuffer(MultiInsertInfo *miinfo,
+ ResultRelInfo *rri)
+{
+ MultiInsertBuffer *buffer;
+
+ buffer = MultiInsertBufferInit(rri);
+
+ /* Setup back-link so we can easily find this buffer again */
+ rri->ri_MultiInsertBuffer = buffer;
+ /* Record that we're tracking this buffer */
+ miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
+}
+
+/*
+ * Initialize an already allocated MultiInsertInfo.
+ *
+ * If rri is a non-partitioned table then a MultiInsertBuffer is set up
+ * for that table.
+ */
+static inline void
+MultiInsertInfoInit(MultiInsertInfo *miinfo, ResultRelInfo *rri,
+ TransitionCaptureState *transition_capture,
+ EState *estate, CommandId mycid, int ti_options)
+{
+ miinfo->multiInsertBuffers = NIL;
+ miinfo->bufferedTuples = 0;
+ miinfo->bufferedBytes = 0;
+ miinfo->transition_capture = transition_capture;
+ miinfo->estate = estate;
+ miinfo->mycid = mycid;
+ miinfo->ti_options = ti_options;
+ miinfo->cur_lineno = 0;
+
+ /*
+ * 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)
+ MultiInsertInfoSetupBuffer(miinfo, rri);
+}
+
+/*
+ * Returns true if the buffers are full
+ */
+static inline bool
+MultiInsertInfoIsFull(MultiInsertInfo *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
+MultiInsertInfoIsEmpty(MultiInsertInfo *miinfo)
+{
+ return miinfo->bufferedTuples == 0;
+}
+
+/*
+ * Write the tuples stored in 'buffer' out to the table.
+ */
+static inline void
+MultiInsertBufferFlush(MultiInsertInfo *miinfo,
+ MultiInsertBuffer *buffer)
+{
+ MemoryContext oldcontext;
+ int i;
+ uint64 save_cur_lineno;
+ EState *estate = miinfo->estate;
+ CommandId mycid = miinfo->mycid;
+ int ti_options = miinfo->ti_options;
+ bool line_buf_valid = miinfo->line_buf_valid;
+ int nused = buffer->nused;
+ ResultRelInfo *resultRelInfo = buffer->resultRelInfo;
+ TupleTableSlot **slots = buffer->slots;
+
+ /*
+ * Print error context information correctly, if one of the operations
+ * below fail.
+ */
+ miinfo->line_buf_valid = false;
+ save_cur_lineno = miinfo->cur_lineno;
+
+ /*
+ * table_multi_insert may leak memory, so switch to short-lived memory
+ * context before calling it.
+ */
+ oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); // XXX requires executor.h
+ 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;
+
+ miinfo->cur_lineno = buffer->linenos[i];
+ recheckIndexes =
+ ExecInsertIndexTuples(resultRelInfo,
+ buffer->slots[i], estate, false, NULL,
+ NIL);
+ ExecARInsertTriggers(estate, resultRelInfo,
+ slots[i], recheckIndexes,
+ miinfo->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))
+ {
+ miinfo->cur_lineno = buffer->linenos[i];
+ ExecARInsertTriggers(estate, resultRelInfo,
+ slots[i], NIL, miinfo->transition_capture);
+ }
+
+ ExecClearTuple(slots[i]);
+ }
+
+ /* Mark that all slots are free */
+ buffer->nused = 0;
+
+ /* reset cur_lineno and line_buf_valid to what they were */
+ miinfo->line_buf_valid = line_buf_valid;
+ miinfo->cur_lineno = save_cur_lineno;
+}
+
+/*
+ * Drop used slots and free member for this buffer.
+ *
+ * The buffer must be flushed before cleanup.
+ */
+static inline void
+MultiInsertBufferCleanup(MultiInsertInfo *miinfo,
+ MultiInsertBuffer *buffer)
+{
+ int i;
+
+ /* Ensure buffer was flushed */
+ Assert(buffer->nused == 0);
+
+ /* Remove back-link to ourself */
+ buffer->resultRelInfo->ri_MultiInsertBuffer = NULL;
+
+ FreeBulkInsertState(buffer->bistate);
+
+ /* 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]);
+
+ table_finish_bulk_insert(buffer->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' is the ResultRelInfo that's currently being
+ * used. When cleaning up old buffers we'll never remove the one for
+ * 'curr_rri'.
+ */
+static inline void
+MultiInsertInfoFlush(MultiInsertInfo *miinfo, ResultRelInfo *curr_rri)
+{
+ ListCell *lc;
+
+ foreach(lc, miinfo->multiInsertBuffers)
+ {
+ MultiInsertBuffer *buffer = (MultiInsertBuffer *) lfirst(lc);
+
+ MultiInsertBufferFlush(miinfo, buffer);
+ }
+
+ 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)
+ {
+ MultiInsertBuffer *buffer;
+
+ buffer = (MultiInsertBuffer *) 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)
+ {
+ miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
+ miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer);
+ buffer = (MultiInsertBuffer *) linitial(miinfo->multiInsertBuffers);
+ }
+
+ MultiInsertBufferCleanup(miinfo, buffer);
+ miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers);
+ }
+}
+
+/*
+ * Cleanup allocated buffers and free memory
+ */
+static inline void
+MultiInsertInfoCleanup(MultiInsertInfo *miinfo)
+{
+ ListCell *lc;
+
+ foreach(lc, miinfo->multiInsertBuffers)
+ MultiInsertBufferCleanup(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 *
+MultiInsertInfoNextFreeSlot(MultiInsertInfo *miinfo,
+ ResultRelInfo *rri)
+{
+ MultiInsertBuffer *buffer = rri->ri_MultiInsertBuffer;
+ int nused = buffer->nused;
+
+ Assert(buffer != NULL);
+ Assert(nused < MAX_BUFFERED_TUPLES);
+
+ 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
+ * MultiInsertInfoNextFreeSlot as being consumed.
+ */
+static inline void
+MultiInsertInfoStore(MultiInsertInfo *miinfo, ResultRelInfo *rri,
+ TupleTableSlot *slot, int tuplen, uint64 lineno)
+{
+ MultiInsertBuffer *buffer = rri->ri_MultiInsertBuffer;
+
+ 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;
+}
+
#endif /* NODEMODIFYTABLE_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 024b0e4da4..ab7b8fb51b 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -33,6 +33,9 @@
#include "utils/tuplesort.h"
#include "utils/tuplestore.h"
+/* This would be a circular inclusion */
+// #include "executor/nodeModifyTable.h"
+
struct PlanState; /* forward references in this file */
struct ParallelHashJoinState;
struct ExecRowMark;
@@ -40,8 +43,8 @@ struct ExprState;
struct ExprContext;
struct RangeTblEntry; /* avoid including parsenodes.h here */
struct ExprEvalStep; /* avoid including execExpr.h everywhere */
-struct CopyMultiInsertBuffer;
-
+// struct MultiInsertBuffer;
+// struct MultiInsertInfo;
/* ----------------
* ExprState node
@@ -499,8 +502,8 @@ typedef struct ResultRelInfo
*/
TupleConversionMap *ri_ChildToRootMap;
- /* for use by copyfrom.c when performing multi-inserts */
- struct CopyMultiInsertBuffer *ri_CopyMultiInsertBuffer;
+ /* for use by copyfrom.c/modifyTable when performing multi-inserts */
+ struct MultiInsertBuffer *ri_MultiInsertBuffer;
} ResultRelInfo;
/* ----------------
@@ -1166,8 +1169,9 @@ typedef struct ModifyTableState
List **mt_arowmarks; /* per-subplan ExecAuxRowMark lists */
EPQState mt_epqstate; /* for evaluating EvalPlanQual rechecks */
bool fireBSTriggers; /* do we need to fire stmt triggers? */
- BulkInsertState bistate; /* state for bulk insert like INSERT SELECT */
+ BulkInsertState bistate; /* state for bulk insert like INSERT SELECT, when miinfo cannot be used */
ResultRelInfo *prevResultRelInfo; /* last child inserted with bistate */
+ struct MultiInsertInfo *miinfo;
/*
* Slot for storing tuples in the root partitioned table's rowtype during
diff --git a/src/test/regress/expected/insert.out b/src/test/regress/expected/insert.out
index da0dae6240..e0c83d7427 100644
--- a/src/test/regress/expected/insert.out
+++ b/src/test/regress/expected/insert.out
@@ -485,6 +485,22 @@ select count(1) from hash_parted;
(1 row)
commit;
+-- test that index was updated
+vacuum analyze hash_parted;
+explain(costs off)
+select * from hash_parted where a=13;
+ QUERY PLAN
+----------------------------------------------------------
+ Index Only Scan using hpart1_a_idx on hpart1 hash_parted
+ Index Cond: (a = 13)
+(2 rows)
+
+select * from hash_parted where a=13;
+ a
+----
+ 13
+(1 row)
+
-- cleanup
drop table range_parted, list_parted;
drop table hash_parted;
diff --git a/src/test/regress/sql/insert.sql b/src/test/regress/sql/insert.sql
index d3a94f053b..99ec18d9a2 100644
--- a/src/test/regress/sql/insert.sql
+++ b/src/test/regress/sql/insert.sql
@@ -292,6 +292,11 @@ select * from hash_parted;
insert into hash_parted select generate_series(1,9999);
select count(1) from hash_parted;
commit;
+-- test that index was updated
+vacuum analyze hash_parted;
+explain(costs off)
+select * from hash_parted where a=13;
+select * from hash_parted where a=13;
-- cleanup
drop table range_parted, list_parted;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index b8ca8cffd9..42415c9c4c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -422,8 +422,6 @@ ConvertRowtypeExpr
CookedConstraint
CopyDest
CopyInsertMethod
-CopyMultiInsertBuffer
-CopyMultiInsertInfo
CopyState
CopyStateData
CopyStmt
@@ -1388,6 +1386,8 @@ ModifyTableState
MorphOpaque
MsgType
MultiAssignRef
+MultiInsertBuffer
+MultiInsertInfo
MultiSortSupport
MultiSortSupportData
MultiXactId
--
2.17.0
--xn9xNsWbHJd/50IB
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v7-0003-Dynamically-switch-to-multi-insert-mode.patch"
^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: Account löschen
@ 2023-03-12 13:22 Alvaro Herrera <[email protected]>
0 siblings, 0 replies; 4+ messages in thread
From: Alvaro Herrera @ 2023-03-12 13:22 UTC (permalink / raw)
To: [email protected]
On 2023-Mar-12, Koray Ili wrote:
> Sehr geehrte Damen und Herren,
>
> ich bitte um Löschung meines Accountes mit dem Benutzername
> [...]
This has been taken care of.
--
Álvaro Herrera
^ permalink raw reply [nested|flat] 4+ messages in thread
end of thread, other threads:[~2023-03-12 13:22 UTC | newest]
Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2020-10-25 03:49 [PATCH v6 2/3] Make INSERT SELECT use multi_insert Justin Pryzby <[email protected]>
2020-10-25 03:49 [PATCH v7 2/3] Make INSERT SELECT use multi_insert Justin Pryzby <[email protected]>
2020-10-25 03:49 [PATCH v5 2/2] Make INSERT SELECT use multi_insert Justin Pryzby <[email protected]>
2023-03-12 13:22 Re: Account löschen Alvaro Herrera <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox