On Mon, Nov 02, 2020 at 12:45:51PM -0600, Justin Pryzby wrote: > On Mon, Nov 02, 2020 at 07:53:45AM +0100, Luc Vlaming wrote: > > On 30.10.20 05:51, Justin Pryzby wrote: > > > On Thu, Oct 22, 2020 at 01:29:53PM +0100, Simon Riggs wrote: > > > > On Fri, 16 Oct 2020 at 22:05, Justin Pryzby <pry...@telsasoft.com> > > > > wrote: > > > > > > > > > > > I made this conditional on BEGIN BULK/SET bulk, so I'll solicit > > > > > > > comments on that. > > > > > > > > I think it would be better if this was self-tuning. So that we don't > > > > allocate a bulkinsert state until we've done say 100 (?) rows > > > > inserted. > > > > > > I made it an optional, non-default behavior in response to the legitimate > > > concern for performance regression for the cases where a loader needs to > > > be as > > > fast as possible - as compared with our case, where we want instead to > > > optimize > > > for our reports by making the loaders responsible for their own writes, > > > rather > > > than leaving behind many dirty pages, and clobbering the cache, too. > > > > > > Also, INSERT SELECT doesn't immediately help us (telsasoft), since we use > > > INSERT .. VALUES () .. ON CONFLICT. This would handle that case, which is > > > great, even though that wasn't a design goal. It could also be an > > > integer GUC > > > to allow configuring the size of the ring buffer. > > > > > > > You should also use table_multi_insert() since that will give further > > > > performance gains by reducing block access overheads. Switching from > > > > single row to multi-row should also only happen once we've loaded a > > > > few rows, so we don't introduce overahads for smaller SQL statements. > > > > > > Good idea...multi_insert (which reduces the overhead of individual > > > inserts) is > > > mostly independent from BulkInsert state (which uses a ring-buffer to > > > avoid > > > dirtying the cache). I made this 0002. > > > > > > This makes INSERT SELECT several times faster, and not clobber the cache > > > too.
- Rebased on Heikki's copy.c split; - Rename structures without "Copy" prefix; - Move MultiInsert* from copyfrom.c to (tentatively) nodeModifyTable.h; - Move cur_lineno and transition_capture into MultiInsertInfo; This switches to multi insert after a configurable number of tuples. If set to -1, that provides the historic behavior that bulk inserts can leave behind many dirty buffers. Perhaps that should be the default. I guess this shouldn't be in copy.h or in commands/* at all. It'll be included by both: commands/copyfrom_internal.h and executor/nodeModifyTable.h. Maybe it should go in util or lib... I don't know how to do it without including executor.h, which seems to be undesirable. -- Justin
>From d67c82e870c640e6f4ba25b3da5acf54df7165d2 Mon Sep 17 00:00:00 2001 From: Justin Pryzby <pryz...@telsasoft.com> Date: Fri, 8 May 2020 02:17:32 -0500 Subject: [PATCH v6 1/3] Allow INSERT SELECT to use a BulkInsertState --- src/backend/executor/nodeModifyTable.c | 22 ++++++++++++++++++++-- src/backend/parser/gram.y | 7 ++++++- src/backend/tcop/utility.c | 4 ++++ src/backend/utils/misc/guc.c | 11 +++++++++++ src/include/executor/nodeModifyTable.h | 2 ++ src/include/nodes/execnodes.h | 3 +++ src/include/parser/kwlist.h | 1 + src/test/regress/expected/insert.out | 23 +++++++++++++++++++++++ src/test/regress/sql/insert.sql | 13 +++++++++++++ 9 files changed, 83 insertions(+), 3 deletions(-) diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index 29e07b7228..26ff964105 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -72,6 +72,8 @@ static TupleTableSlot *ExecPrepareTupleRouting(ModifyTableState *mtstate, ResultRelInfo *targetRelInfo, TupleTableSlot *slot, ResultRelInfo **partRelInfo); +/* guc */ +bool insert_in_bulk = false; /* * Verify that the tuples to be produced by INSERT or UPDATE match the @@ -594,7 +596,7 @@ ExecInsert(ModifyTableState *mtstate, table_tuple_insert_speculative(resultRelationDesc, slot, estate->es_output_cid, 0, - NULL, + NULL, /* Bulk insert not supported */ specToken); /* insert index entries for tuple */ @@ -631,10 +633,17 @@ ExecInsert(ModifyTableState *mtstate, } else { + if (proute && mtstate->prevResultRelInfo != resultRelInfo) + { + if (mtstate->bistate) + ReleaseBulkInsertStatePin(mtstate->bistate); + mtstate->prevResultRelInfo = resultRelInfo; + } + /* insert the tuple normally */ table_tuple_insert(resultRelationDesc, slot, estate->es_output_cid, - 0, NULL); + 0, mtstate->bistate); /* insert index entries for tuple */ if (resultRelInfo->ri_NumIndices > 0) @@ -2232,6 +2241,9 @@ 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) + mtstate->bistate = GetBulkInsertState(); /* set up epqstate with dummy subplan data for the moment */ EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, node->epqParam); @@ -2698,6 +2710,12 @@ ExecEndModifyTable(ModifyTableState *node) resultRelInfo); } + if (node->bistate) + { + FreeBulkInsertState(node->bistate); + table_finish_bulk_insert(node->rootResultRelInfo->ri_RelationDesc, 0); + } + /* * Close all the partitioned tables, leaf partitions, and their indices * and release the slot used for tuple routing, if set. diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index efc9c99754..5915c8c414 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -631,7 +631,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); ASSERTION ASSIGNMENT ASYMMETRIC AT ATTACH ATTRIBUTE AUTHORIZATION BACKWARD BEFORE BEGIN_P BETWEEN BIGINT BINARY BIT - BOOLEAN_P BOTH BY + BOOLEAN_P BOTH BULK BY CACHE CALL CALLED CASCADE CASCADED CASE CAST CATALOG_P CHAIN CHAR_P CHARACTER CHARACTERISTICS CHECK CHECKPOINT CLASS CLOSE @@ -9886,6 +9886,9 @@ transaction_mode_item: | NOT DEFERRABLE { $$ = makeDefElem("transaction_deferrable", makeIntConst(false, @1), @1); } + | BULK + { $$ = makeDefElem("bulk", + makeIntConst(true, @1), @1); } ; /* Syntax with commas is SQL-spec, without commas is Postgres historical */ @@ -15157,6 +15160,7 @@ unreserved_keyword: | BACKWARD | BEFORE | BEGIN_P + | BULK | BY | CACHE | CALL @@ -15668,6 +15672,7 @@ bare_label_keyword: | BIT | BOOLEAN_P | BOTH + | BULK | BY | CACHE | CALL diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 81ac9b1cb2..a0a4034409 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -611,6 +611,10 @@ standard_ProcessUtility(PlannedStmt *pstmt, SetPGVariable("transaction_deferrable", list_make1(item->arg), true); + else if (strcmp(item->defname, "bulk") == 0) + SetPGVariable("bulk_insert", + list_make1(item->arg), + true); } } break; diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index bb34630e8e..170c96e749 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -54,6 +54,7 @@ #include "libpq/pqformat.h" #include "miscadmin.h" #include "optimizer/cost.h" +#include "executor/nodeModifyTable.h" #include "optimizer/geqo.h" #include "optimizer/optimizer.h" #include "optimizer/paths.h" @@ -2036,6 +2037,16 @@ static struct config_bool ConfigureNamesBool[] = NULL, NULL, NULL }, + { + {"bulk_insert", PGC_USERSET, CLIENT_CONN_STATEMENT, + gettext_noop("Sets the transaction to bulk insert mode."), + gettext_noop("A ring buffer of limited size will be used."), + }, + &insert_in_bulk, + false, + NULL, NULL, NULL + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, false, NULL, NULL, NULL diff --git a/src/include/executor/nodeModifyTable.h b/src/include/executor/nodeModifyTable.h index 46a2dc9511..09c312a052 100644 --- a/src/include/executor/nodeModifyTable.h +++ b/src/include/executor/nodeModifyTable.h @@ -15,6 +15,8 @@ #include "nodes/execnodes.h" +extern PGDLLIMPORT bool insert_in_bulk; + extern void ExecComputeStoredGenerated(ResultRelInfo *resultRelInfo, EState *estate, TupleTableSlot *slot, CmdType cmdtype); diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index f6824bf2e1..c79d13aa44 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -14,6 +14,7 @@ #ifndef EXECNODES_H #define EXECNODES_H +#include "access/heapam.h" #include "access/tupconvert.h" #include "executor/instrument.h" #include "fmgr.h" @@ -1176,6 +1177,8 @@ 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 */ + ResultRelInfo *prevResultRelInfo; /* last child inserted with bistate */ /* * Slot for storing tuples in the root partitioned table's rowtype during diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h index 71dcdf2889..0991da11e7 100644 --- a/src/include/parser/kwlist.h +++ b/src/include/parser/kwlist.h @@ -60,6 +60,7 @@ PG_KEYWORD("binary", BINARY, TYPE_FUNC_NAME_KEYWORD, BARE_LABEL) PG_KEYWORD("bit", BIT, COL_NAME_KEYWORD, BARE_LABEL) PG_KEYWORD("boolean", BOOLEAN_P, COL_NAME_KEYWORD, BARE_LABEL) PG_KEYWORD("both", BOTH, RESERVED_KEYWORD, BARE_LABEL) +PG_KEYWORD("bulk", BULK, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("by", BY, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("cache", CACHE, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("call", CALL, UNRESERVED_KEYWORD, BARE_LABEL) diff --git a/src/test/regress/expected/insert.out b/src/test/regress/expected/insert.out index da50ee3b67..da0dae6240 100644 --- a/src/test/regress/expected/insert.out +++ b/src/test/regress/expected/insert.out @@ -462,6 +462,29 @@ Partitions: part_aa_bb FOR VALUES IN ('aa', 'bb'), part_xx_yy FOR VALUES IN ('xx', 'yy'), PARTITIONED, part_default DEFAULT, PARTITIONED +-- bulk inserts +truncate hash_parted; +begin bulk; +create index on hash_parted(a); +-- make sure small inserts are flushed +insert into hash_parted values(11); +insert into hpart0 values(12); +select * from hash_parted; + a +---- + 12 + 11 +(2 rows) + +-- exercise bulk insert to partitions +insert into hash_parted select generate_series(1,9999); +select count(1) from hash_parted; + count +------- + 10001 +(1 row) + +commit; -- 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 963faa1614..d3a94f053b 100644 --- a/src/test/regress/sql/insert.sql +++ b/src/test/regress/sql/insert.sql @@ -280,6 +280,19 @@ from hash_parted order by part; -- partitions \d+ list_parted +-- bulk inserts +truncate hash_parted; +begin bulk; +create index on hash_parted(a); +-- make sure small inserts are flushed +insert into hash_parted values(11); +insert into hpart0 values(12); +select * from hash_parted; +-- exercise bulk insert to partitions +insert into hash_parted select generate_series(1,9999); +select count(1) from hash_parted; +commit; + -- cleanup drop table range_parted, list_parted; drop table hash_parted; -- 2.17.0
>From 6aee47afb4e0f0730dc03398782ec102a5253aad Mon Sep 17 00:00:00 2001 From: Justin Pryzby <pryz...@telsasoft.com> Date: Sat, 24 Oct 2020 22:49:01 -0500 Subject: [PATCH v6 2/3] Make INSERT SELECT use multi_insert 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
>From 1605d4b12cc638ebcb80c044f0bc7ff55e119fa4 Mon Sep 17 00:00:00 2001 From: Justin Pryzby <pryz...@telsasoft.com> Date: Mon, 23 Nov 2020 17:57:24 -0600 Subject: [PATCH v6 3/3] Dynamically switch to multi-insert mode.. by popular request --- src/backend/executor/nodeModifyTable.c | 47 +++++++++++++++----------- src/backend/tcop/utility.c | 4 --- src/backend/utils/misc/guc.c | 19 +++++------ src/include/executor/nodeModifyTable.h | 3 +- src/include/nodes/execnodes.h | 1 + src/test/regress/expected/insert.out | 4 +++ src/test/regress/sql/insert.sql | 2 ++ 7 files changed, 46 insertions(+), 34 deletions(-) diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index f8cd537b24..96d5a5d164 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -74,7 +74,7 @@ static TupleTableSlot *ExecPrepareTupleRouting(ModifyTableState *mtstate, TupleTableSlot *slot, ResultRelInfo **partRelInfo); /* guc */ -bool insert_in_bulk = false; +int bulk_insert_ntuples = 1000; /* * Verify that the tuples to be produced by INSERT or UPDATE match the @@ -413,6 +413,33 @@ ExecInsert(ModifyTableState *mtstate, resultRelationDesc = resultRelInfo->ri_RelationDesc; + /* Use bulk insert after a threshold number of tuples */ + mtstate->ntuples++; + if (mtstate->bistate == NULL && + mtstate->operation == CMD_INSERT && + onconflict == ONCONFLICT_NONE && + mtstate->ntuples > bulk_insert_ntuples && + bulk_insert_ntuples >= 0) + { + elog(DEBUG1, "enabling bulk insert"); + mtstate->bistate = GetBulkInsertState(); + + /* + * For partitioned tables we don't support multi-inserts when there + * are any statement level insert triggers. + */ + if (proute == 0 || + !mtstate->rootResultRelInfo->ri_TrigDesc || + !mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_new_table) + { + elog(DEBUG1, "enabling multi insert"); + mtstate->miinfo = calloc(sizeof(*mtstate->miinfo), 1); + MultiInsertInfoInit(mtstate->miinfo, mtstate->rootResultRelInfo, + mtstate->mt_transition_capture, + estate, GetCurrentCommandId(true), 0); + } + } + /* * BEFORE ROW INSERT Triggers. * @@ -2285,24 +2312,6 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) 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); diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index a0a4034409..81ac9b1cb2 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -611,10 +611,6 @@ standard_ProcessUtility(PlannedStmt *pstmt, SetPGVariable("transaction_deferrable", list_make1(item->arg), true); - else if (strcmp(item->defname, "bulk") == 0) - SetPGVariable("bulk_insert", - list_make1(item->arg), - true); } } break; diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 170c96e749..3ed6b0650c 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -2037,16 +2037,6 @@ static struct config_bool ConfigureNamesBool[] = NULL, NULL, NULL }, - { - {"bulk_insert", PGC_USERSET, CLIENT_CONN_STATEMENT, - gettext_noop("Sets the transaction to bulk insert mode."), - gettext_noop("A ring buffer of limited size will be used."), - }, - &insert_in_bulk, - false, - NULL, NULL, NULL - }, - /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, false, NULL, NULL, NULL @@ -3410,6 +3400,15 @@ static struct config_int ConfigureNamesInt[] = check_huge_page_size, NULL, NULL }, + { + {"bulk_insert_ntuples", PGC_USERSET, CLIENT_CONN_STATEMENT, + gettext_noop("Enable bulk insertions after this number of tuples."), + gettext_noop("A ring buffer of limited size will be used and updates done in batch"), + }, + &bulk_insert_ntuples, + 1000, -1, INT_MAX, + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL diff --git a/src/include/executor/nodeModifyTable.h b/src/include/executor/nodeModifyTable.h index 80d82a9715..16a04151c8 100644 --- a/src/include/executor/nodeModifyTable.h +++ b/src/include/executor/nodeModifyTable.h @@ -17,7 +17,7 @@ #include "executor/executor.h" // XXX #include "nodes/execnodes.h" -extern PGDLLIMPORT bool insert_in_bulk; +extern PGDLLIMPORT int bulk_insert_ntuples; extern void ExecComputeStoredGenerated(ResultRelInfo *resultRelInfo, EState *estate, TupleTableSlot *slot, @@ -74,6 +74,7 @@ typedef struct MultiInsertInfo EState *estate; /* Executor state used for COPY */ CommandId mycid; /* Command Id used for COPY */ int ti_options; /* table insert options */ + // size_t ntuples; /* Number of rows inserted */ /* Line number for errors in copyfrom.c */ uint64 cur_lineno; diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 985a98f854..535380d276 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -1183,6 +1183,7 @@ typedef struct ModifyTableState BulkInsertState bistate; /* state for bulk insert like INSERT SELECT, when miinfo cannot be used */ ResultRelInfo *prevResultRelInfo; /* last child inserted with bistate */ struct MultiInsertInfo *miinfo; + size_t ntuples; /* Number of tuples inserted; XXX: could be in 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 e0c83d7427..b894180152 100644 --- a/src/test/regress/expected/insert.out +++ b/src/test/regress/expected/insert.out @@ -477,7 +477,11 @@ select * from hash_parted; (2 rows) -- exercise bulk insert to partitions +SET client_min_messages=debug; insert into hash_parted select generate_series(1,9999); +DEBUG: enabling bulk insert +DEBUG: enabling multi insert +RESET client_min_messages; select count(1) from hash_parted; count ------- diff --git a/src/test/regress/sql/insert.sql b/src/test/regress/sql/insert.sql index 99ec18d9a2..ff9b57af5e 100644 --- a/src/test/regress/sql/insert.sql +++ b/src/test/regress/sql/insert.sql @@ -289,7 +289,9 @@ insert into hash_parted values(11); insert into hpart0 values(12); select * from hash_parted; -- exercise bulk insert to partitions +SET client_min_messages=debug; insert into hash_parted select generate_series(1,9999); +RESET client_min_messages; select count(1) from hash_parted; commit; -- test that index was updated -- 2.17.0