public inbox for [email protected]
help / color / mirror / Atom feedFrom: Justin Pryzby <[email protected]>
Subject: [PATCH v5 2/2] Make INSERT SELECT use multi_insert
Date: Sat, 24 Oct 2020 22:49:01 -0500
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--
view thread (12+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH v5 2/2] Make INSERT SELECT use multi_insert
In-Reply-To: <no-message-id-1864637@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox