Hi, On Thu, May 16, 2024 at 5:01 AM Jeff Davis <pg...@j-davis.com> wrote: > > The flushing behavior is entirely controlled by the table AM. The heap > can use the same flushing logic that it did before, which is to hold > 1000 tuples. > > I like that it's accounting for memory, too, but it doesn't need to be > overly restrictive. Why not just use work_mem? That should hold 1000 > reasonably-sized tuples, plus overhead. > > Even better would be if we could take into account partitioning. That > might be out of scope for your current work, but it would be very > useful. We could have a couple new GUCs like modify_table_buffer and > modify_table_buffer_per_partition or something like that.
I disagree with inventing more GUCs. Instead, I'd vote for just holding 1000 tuples in buffers for heap AM. This not only keeps the code and new table AM simple, but also does not cause regression for COPY. In my testing, 1000 tuples with 1 int and 1 float columns took 40000 bytes of memory (40 bytes each tuple), whereas with 1 int, 1 float and 1 text columns took 172000 bytes of memory (172 bytes each tuple) bytes which IMO mustn't be a big problem. Thoughts? > > 1. Try to get the actual tuple sizes excluding header sizes for each > > column in the new TAM. > > I don't see the point in arbitrarily excluding the header. > > > v21 also adds code to maintain tuple size for virtual tuple slots. > > This helps make better memory-based flushing decisions in the new > > TAM. > > That seems wrong. We shouldn't need to change the TupleTableSlot > structure for this patch. I dropped these ideas as I went ahead with the above idea of just holding 1000 tuples in buffers for heap AM. > Comments on v21: > > * All callers specify TM_FLAG_MULTI_INSERTS. What's the purpose? Previously, the multi insert state was initialized in modify_begin, so it was then required to differentiate the code path. But, it's not needed anymore with the lazy initialization of the multi insert state moved to modify_buffer_insert. I removed it. > * The only caller that doesn't use TM_FLAG_BAS_BULKWRITE is > ExecInsert(). What's the disadvantage to using a bulk insert state > there? The subsequent read queries will not find the just-now-inserted tuples in shared buffers as a separate ring buffer is used with bulk insert access strategy. The multi inserts is nothing but buffering multiple tuples plus inserting in bulk. So using the bulk insert strategy might be worth it for INSERT INTO SELECTs too. Thoughts? > * I'm a bit confused by TableModifyState->modify_end_callback. The AM > both sets the callback and calls the callback -- why can't the code > just go into the table_modify_end method? I came up with modify_end_callback as per the discussion upthread to use modify_begin, modify_end in future for UPDATE, DELETE and MERGE, and not use any operation specific flags to clean the state appropriately. The operation specific state cleaning logic can go to the modify_end_callback implementation defined by the AM. > * The code structure in table_modify_begin() (and related) is strange. > Can it be simplified or am I missing something? I previously defined these new table AMs as optional, check GetTableAmRoutine(). And, there was a point upthread to provide default/fallback implementation to help not fail insert operations on tables without the new table AMs implemented. FWIW, the default implementation was just doing the single inserts. The table_modify_begin and friends need the logic to fallback making the code there look different than other AMs. However, I now have a feeling to drop the idea of having fallback implementation and let the AMs deal with it. Although it might create some friction with various non-core AM implementations, it keeps this patch simple which I would vote for. Thoughts? > * Why are table_modify_state and insert_modify_buffer_flush_context > globals? What if there are multiple modify nodes in a plan? Can you please provide the case that can generate multiple "modify nodes" in a single plan? AFAICS, multiple "modify nodes" in a plan can exist for both partitioned tables and tables that get created as part of CTEs. I disabled multi inserts for both of these cases. The way I disabled for CTEs looks pretty naive - I just did the following. Any better suggestions here to deal with all such cases? + if (operation == CMD_INSERT && + nodeTag(subplanstate) == T_SeqScanState) + canMultiInsert = true; > * Can you explain the design in logical rep? Multi inserts for logical replication work at the table level. In other words, all tuple inserts related to a single table within a transaction are buffered and written to the corresponding table when necessary. Whenever inserts pertaining to another table arrive, the buffered tuples related to the previous table are written to the table before starting the buffering for the new table. Also, the tuples are written to the table from the buffer when there arrives a non-INSERT operation, for example, UPDATE/DELETE/TRUNCATE/COMMIT etc. FWIW, pglogical has the similar multi inserts logic - https://github.com/2ndQuadrant/pglogical/blob/REL2_x_STABLE/pglogical_apply_heap.c#L879. Please find the v22 patches with the above changes. -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com
From ee6714e79bdcfe3d0e104f84caee8bd4be730ddb Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com> Date: Tue, 4 Jun 2024 20:02:52 +0000 Subject: [PATCH v22 1/5] Introduce new Table AM for multi inserts Until now, it's the COPY ... FROM command using multi inserts (i.e. buffer some tuples and inserts them to table at once). Various other commands can benefit from this multi insert logic [Reusable]. Also, there's a need to have these multi insert AMs (Access Methods) as scan-like API [Usability]. Also, there's a need allow various table AMs define their own buffering and flushing strategy [Flexibility]. This commit introduces, new table AMs for multi inserts to help achieve all of the above. Upcoming commits will have these new table AMs being used for various other commands. Author: Bharath Rupireddy Reviewed-by: Jeff Davis Discussion: https://www.postgresql.org/message-id/20200924024128.kyk3r5g7dnu3fxxx@alap3.anarazel.de Discussion: https://www.postgresql.org/message-id/CALj2ACVi9eTRYR%3Dgdca5wxtj3Kk_9q9qVccxsS1hngTGOCjPwQ%40mail.gmail.com Discussion: https://www.postgresql.org/message-id/8633171cb034aafc260fdf37df04b6c779aa1e2f.camel%40j-davis.com --- src/backend/access/heap/heapam.c | 197 ++++++++++++++++++++++- src/backend/access/heap/heapam_handler.c | 6 + src/backend/access/table/tableamapi.c | 5 + src/include/access/heapam.h | 38 +++++ src/include/access/tableam.h | 80 +++++++++ src/tools/pgindent/typedefs.list | 3 + 6 files changed, 328 insertions(+), 1 deletion(-) diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 82bb9cb33b..aaf7a26389 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -64,6 +64,7 @@ #include "storage/standby.h" #include "utils/datum.h" #include "utils/inval.h" +#include "utils/memutils.h" #include "utils/relcache.h" #include "utils/snapmgr.h" #include "utils/spccache.h" @@ -112,7 +113,7 @@ static int bottomup_sort_and_shrink(TM_IndexDeleteOp *delstate); static XLogRecPtr log_heap_new_cid(Relation relation, HeapTuple tup); static HeapTuple ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_required, bool *copy); - +static void heap_modify_insert_end_callback(TableModifyState *state); /* * Each tuple lock mode has a corresponding heavyweight lock, and one or two @@ -2611,6 +2612,200 @@ heap_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples, pgstat_count_heap_insert(relation, ntuples); } +/* + * Initialize heap modify state. + */ +TableModifyState * +heap_modify_begin(Relation rel, int modify_flags, + CommandId cid, int options, + TableModifyBufferFlushCallback modify_buffer_flush_callback, + void *modify_buffer_flush_context) +{ + TableModifyState *state; + MemoryContext context; + MemoryContext oldcontext; + + context = AllocSetContextCreate(TopTransactionContext, + "heap_modify memory context", + ALLOCSET_DEFAULT_SIZES); + + oldcontext = MemoryContextSwitchTo(context); + state = palloc0(sizeof(TableModifyState)); + state->rel = rel; + state->modify_flags = modify_flags; + state->mem_cxt = context; + state->cid = cid; + state->options = options; + state->modify_buffer_flush_callback = modify_buffer_flush_callback; + state->modify_buffer_flush_context = modify_buffer_flush_context; + state->modify_end_callback = NULL; /* To be installed lazily */ + MemoryContextSwitchTo(oldcontext); + + return state; +} + +/* + * Store passed-in tuple into in-memory buffered slots. When full, insert + * multiple tuples from the buffers into heap. + */ +void +heap_modify_buffer_insert(TableModifyState *state, + TupleTableSlot *slot) +{ + TupleTableSlot *dstslot; + HeapInsertState *istate; + HeapMultiInsertState *mistate; + MemoryContext oldcontext; + + oldcontext = MemoryContextSwitchTo(state->mem_cxt); + + /* First time through, initialize heap insert state */ + if (state->data == NULL) + { + istate = (HeapInsertState *) palloc0(sizeof(HeapInsertState)); + istate->bistate = NULL; + istate->mistate = NULL; + state->data = istate; + mistate = (HeapMultiInsertState *) palloc0(sizeof(HeapMultiInsertState)); + mistate->slots = (TupleTableSlot **) palloc0(sizeof(TupleTableSlot *) * HEAP_MAX_BUFFERED_SLOTS); + istate->mistate = mistate; + mistate->mem_cxt = AllocSetContextCreate(CurrentMemoryContext, + "heap_multi_insert memory context", + ALLOCSET_DEFAULT_SIZES); + + if ((state->modify_flags & TM_FLAG_BAS_BULKWRITE) != 0) + istate->bistate = GetBulkInsertState(); + + state->modify_end_callback = heap_modify_insert_end_callback; + } + + istate = (HeapInsertState *) state->data; + Assert(istate->mistate != NULL); + mistate = istate->mistate; + + dstslot = mistate->slots[mistate->cur_slots]; + if (dstslot == NULL) + { + /* + * We use virtual tuple slots buffered slots for leveraging the + * optimization it provides to minimize physical data copying. The + * virtual slot gets materialized when we copy (via below + * ExecCopySlot) the tuples from the source slot which can be of any + * type. This way, it is ensured that the tuple storage doesn't depend + * on external memory, because all the datums that aren't passed by + * value are copied into the slot's memory context. + */ + dstslot = MakeTupleTableSlot(RelationGetDescr(state->rel), + &TTSOpsVirtual); + mistate->slots[mistate->cur_slots] = dstslot; + } + + Assert(TTS_IS_VIRTUAL(dstslot)); + + /* + * Note that the copy clears the previous destination slot contents, so + * there's no need of explicit ExecClearTuple here. + */ + ExecCopySlot(dstslot, slot); + + mistate->cur_slots++; + + if (mistate->cur_slots >= HEAP_MAX_BUFFERED_SLOTS) + heap_modify_buffer_flush(state); + + MemoryContextSwitchTo(oldcontext); +} + +/* + * Insert multiple tuples from in-memory buffered slots into heap. + */ +void +heap_modify_buffer_flush(TableModifyState *state) +{ + HeapInsertState *istate; + HeapMultiInsertState *mistate; + MemoryContext oldcontext; + + /* Quick exit if we haven't inserted anything yet */ + if (state->data == NULL) + return; + + istate = (HeapInsertState *) state->data; + Assert(istate->mistate != NULL); + mistate = istate->mistate; + + /* Quick exit if we have flushed already */ + if (mistate->cur_slots == 0) + return; + + /* + * heap_multi_insert may leak memory, so switch to short-lived memory + * context before calling it. + */ + oldcontext = MemoryContextSwitchTo(mistate->mem_cxt); + + heap_multi_insert(state->rel, mistate->slots, mistate->cur_slots, + state->cid, state->options, istate->bistate); + + MemoryContextSwitchTo(oldcontext); + MemoryContextReset(mistate->mem_cxt); + + if (state->modify_buffer_flush_callback != NULL) + { + for (int i = 0; i < mistate->cur_slots; i++) + state->modify_buffer_flush_callback(state->modify_buffer_flush_context, + mistate->slots[i]); + } + + mistate->cur_slots = 0; +} + +/* + * Heap insert specific callback used for performing work at the end like + * flushing buffered tuples if any, cleaning up the insert state and buffered + * slots. + */ +static void +heap_modify_insert_end_callback(TableModifyState *state) +{ + HeapInsertState *istate; + + /* Quick exit if we haven't inserted anything yet */ + if (state->data == NULL) + return; + + istate = (HeapInsertState *) state->data; + + if (istate->mistate != NULL) + { + HeapMultiInsertState *mistate = istate->mistate; + + heap_modify_buffer_flush(state); + + Assert(mistate->cur_slots == 0); + + for (int i = 0; i < HEAP_MAX_BUFFERED_SLOTS && mistate->slots[i] != NULL; i++) + ExecDropSingleTupleTableSlot(mistate->slots[i]); + + MemoryContextDelete(mistate->mem_cxt); + } + + if (istate->bistate != NULL) + FreeBulkInsertState(istate->bistate); +} + +/* + * Clean heap modify state. + */ +void +heap_modify_end(TableModifyState *state) +{ + if (state->modify_end_callback != NULL) + state->modify_end_callback(state); + + MemoryContextDelete(state->mem_cxt); +} + /* * simple_heap_insert - insert a tuple * diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 6f8b1b7929..eda0c73a16 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -2615,6 +2615,12 @@ static const TableAmRoutine heapam_methods = { .tuple_insert_speculative = heapam_tuple_insert_speculative, .tuple_complete_speculative = heapam_tuple_complete_speculative, .multi_insert = heap_multi_insert, + + .tuple_modify_begin = heap_modify_begin, + .tuple_modify_buffer_insert = heap_modify_buffer_insert, + .tuple_modify_buffer_flush = heap_modify_buffer_flush, + .tuple_modify_end = heap_modify_end, + .tuple_delete = heapam_tuple_delete, .tuple_update = heapam_tuple_update, .tuple_lock = heapam_tuple_lock, diff --git a/src/backend/access/table/tableamapi.c b/src/backend/access/table/tableamapi.c index e9b598256f..772f29b1b5 100644 --- a/src/backend/access/table/tableamapi.c +++ b/src/backend/access/table/tableamapi.c @@ -97,6 +97,11 @@ GetTableAmRoutine(Oid amhandler) Assert(routine->scan_sample_next_block != NULL); Assert(routine->scan_sample_next_tuple != NULL); + Assert(routine->tuple_modify_begin != NULL); + Assert(routine->tuple_modify_buffer_insert != NULL); + Assert(routine->tuple_modify_buffer_flush != NULL); + Assert(routine->tuple_modify_end != NULL); + return routine; } diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h index c47a5045ce..36ea3d5d2c 100644 --- a/src/include/access/heapam.h +++ b/src/include/access/heapam.h @@ -271,6 +271,32 @@ typedef enum PRUNE_VACUUM_CLEANUP, /* VACUUM 2nd heap pass */ } PruneReason; +/* + * Maximum number of slots that multi-insert buffers can hold. + * + * Caution: Don't make this too big, as we could end up with this many tuples + * stored in multi insert buffer. + */ +#define HEAP_MAX_BUFFERED_SLOTS 1000 + +typedef struct HeapMultiInsertState +{ + /* Array of buffered slots */ + TupleTableSlot **slots; + + /* Number of buffered slots currently held */ + int cur_slots; + + MemoryContext mem_cxt; +} HeapMultiInsertState; + +typedef struct HeapInsertState +{ + struct BulkInsertStateData *bistate; + HeapMultiInsertState *mistate; +} HeapInsertState; + + /* ---------------- * function prototypes for heap access method * @@ -321,6 +347,18 @@ extern void heap_insert(Relation relation, HeapTuple tup, CommandId cid, extern void heap_multi_insert(Relation relation, struct TupleTableSlot **slots, int ntuples, CommandId cid, int options, BulkInsertState bistate); + +extern TableModifyState *heap_modify_begin(Relation rel, + int modify_flags, + CommandId cid, + int options, + TableModifyBufferFlushCallback modify_buffer_flush_callback, + void *modify_buffer_flush_context); +extern void heap_modify_buffer_insert(TableModifyState *state, + TupleTableSlot *slot); +extern void heap_modify_buffer_flush(TableModifyState *state); +extern void heap_modify_end(TableModifyState *state); + extern TM_Result heap_delete(Relation relation, ItemPointer tid, CommandId cid, Snapshot crosscheck, bool wait, struct TM_FailureData *tmfd, bool changingPart); diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h index 8e583b45cd..9e9b9771de 100644 --- a/src/include/access/tableam.h +++ b/src/include/access/tableam.h @@ -255,6 +255,39 @@ typedef struct TM_IndexDeleteOp TM_IndexStatus *status; } TM_IndexDeleteOp; +/* Table modify flags */ + +/* Use BAS_BULKWRITE buffer access strategy */ +#define TM_FLAG_BAS_BULKWRITE 0x000001 + +struct TableModifyState; + +/* Callback invoked for each tuple that gets flushed to disk from buffer */ +typedef void (*TableModifyBufferFlushCallback) (void *context, + TupleTableSlot *slot); + +/* Table AM specific callback that gets called in table_modify_end() */ +typedef void (*TableModifyEndCallback) (struct TableModifyState *state); + +/* Holds table modify state */ +typedef struct TableModifyState +{ + Relation rel; + int modify_flags; + MemoryContext mem_cxt; + CommandId cid; + int options; + + /* Flush callback and its context */ + TableModifyBufferFlushCallback modify_buffer_flush_callback; + void *modify_buffer_flush_context; + + /* Table AM specific data */ + void *data; + + TableModifyEndCallback modify_end_callback; +} TableModifyState; + /* "options" flag bits for table_tuple_insert */ /* TABLE_INSERT_SKIP_WAL was 0x0001; RelationNeedsWAL() now governs */ #define TABLE_INSERT_SKIP_FSM 0x0002 @@ -578,6 +611,21 @@ typedef struct TableAmRoutine void (*finish_bulk_insert) (Relation rel, int options); + /* ------------------------------------------------------------------------ + * Table Modify related functions. + * ------------------------------------------------------------------------ + */ + TableModifyState *(*tuple_modify_begin) (Relation rel, + int modify_flags, + CommandId cid, + int options, + TableModifyBufferFlushCallback modify_buffer_flush_callback, + void *modify_buffer_flush_context); + void (*tuple_modify_buffer_insert) (TableModifyState *state, + TupleTableSlot *slot); + void (*tuple_modify_buffer_flush) (TableModifyState *state); + void (*tuple_modify_end) (TableModifyState *state); + /* ------------------------------------------------------------------------ * DDL related functionality. * ------------------------------------------------------------------------ @@ -1609,6 +1657,38 @@ table_finish_bulk_insert(Relation rel, int options) rel->rd_tableam->finish_bulk_insert(rel, options); } +/* ------------------------------------------------------------------------ + * Table Modify related functions. + * ------------------------------------------------------------------------ + */ +static inline TableModifyState * +table_modify_begin(Relation rel, int modify_flags, CommandId cid, int options, + TableModifyBufferFlushCallback modify_buffer_flush_callback, + void *modify_buffer_flush_context) +{ + return rel->rd_tableam->tuple_modify_begin(rel, modify_flags, + cid, options, + modify_buffer_flush_callback, + modify_buffer_flush_context); +} + +static inline void +table_modify_buffer_insert(TableModifyState *state, TupleTableSlot *slot) +{ + state->rel->rd_tableam->tuple_modify_buffer_insert(state, slot); +} + +static inline void +table_modify_buffer_flush(TableModifyState *state) +{ + state->rel->rd_tableam->tuple_modify_buffer_flush(state); +} + +static inline void +table_modify_end(TableModifyState *state) +{ + state->rel->rd_tableam->tuple_modify_end(state); +} /* ------------------------------------------------------------------------ * DDL related functionality. diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index d427a1c16a..84baf9b78d 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1141,6 +1141,8 @@ HeadlineJsonState HeadlineParsedText HeadlineWordEntry HeapCheckContext +HeapInsertState +HeapMultiInsertState HeapPageFreeze HeapScanDesc HeapTuple @@ -2859,6 +2861,7 @@ TableFuncScanState TableFuncType TableInfo TableLikeClause +TableModifyState TableSampleClause TableScanDesc TableScanDescData -- 2.34.1
From e029a166d6bc7550d544f93817c9c1b6d4aa368e Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com> Date: Tue, 4 Jun 2024 20:17:45 +0000 Subject: [PATCH v22 2/5] Optimize various SQL commands with new multi insert table AM This commit optimizes the following commands for heap AM using new multi insert table AM added by commit <<CHANGE_ME>>: - CREATE TABLE AS - CREATE MATERIALIZED VIEW - REFRESH MATERIALIZED VIEW - Table Rewrites Author: Bharath Rupireddy Reviewed-by: Jeff Davis Discussion: https://www.postgresql.org/message-id/CALj2ACVi9eTRYR%3Dgdca5wxtj3Kk_9q9qVccxsS1hngTGOCjPwQ%40mail.gmail.com Discussion: https://www.postgresql.org/message-id/8633171cb034aafc260fdf37df04b6c779aa1e2f.camel%40j-davis.com --- src/backend/commands/createas.c | 26 ++++++++++---------------- src/backend/commands/matview.c | 25 ++++++++++--------------- src/backend/commands/tablecmds.c | 30 ++++++++++-------------------- 3 files changed, 30 insertions(+), 51 deletions(-) diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c index 62050f4dc5..794c735b78 100644 --- a/src/backend/commands/createas.c +++ b/src/backend/commands/createas.c @@ -53,9 +53,7 @@ typedef struct /* These fields are filled by intorel_startup: */ Relation rel; /* relation to write to */ ObjectAddress reladdr; /* address of rel, for ExecCreateTableAs */ - CommandId output_cid; /* cmin to insert in output tuples */ - int ti_options; /* table_tuple_insert performance options */ - BulkInsertState bistate; /* bulk insert state */ + TableModifyState *mstate; /* table insert state */ } DR_intorel; /* utility functions for CTAS definition creation */ @@ -552,17 +550,20 @@ intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo) */ myState->rel = intoRelationDesc; myState->reladdr = intoRelationAddr; - myState->output_cid = GetCurrentCommandId(true); - myState->ti_options = TABLE_INSERT_SKIP_FSM; /* * If WITH NO DATA is specified, there is no need to set up the state for * bulk inserts as there are no tuples to insert. */ if (!into->skipData) - myState->bistate = GetBulkInsertState(); + myState->mstate = table_modify_begin(intoRelationDesc, + TM_FLAG_BAS_BULKWRITE, + GetCurrentCommandId(true), + TABLE_INSERT_SKIP_FSM, + NULL, + NULL); else - myState->bistate = NULL; + myState->mstate = NULL; /* * Valid smgr_targblock implies something already wrote to the relation. @@ -590,11 +591,7 @@ intorel_receive(TupleTableSlot *slot, DestReceiver *self) * would not be cheap either. This also doesn't allow accessing per-AM * data (say a tuple's xmin), but since we don't do that here... */ - table_tuple_insert(myState->rel, - slot, - myState->output_cid, - myState->ti_options, - myState->bistate); + table_modify_buffer_insert(myState->mstate, slot); } /* We know this is a newly created relation, so there are no indexes */ @@ -612,10 +609,7 @@ intorel_shutdown(DestReceiver *self) IntoClause *into = myState->into; if (!into->skipData) - { - FreeBulkInsertState(myState->bistate); - table_finish_bulk_insert(myState->rel, myState->ti_options); - } + table_modify_end(myState->mstate); /* close rel, but keep lock until commit */ table_close(myState->rel, NoLock); diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c index 6d09b75556..c23861c505 100644 --- a/src/backend/commands/matview.c +++ b/src/backend/commands/matview.c @@ -48,9 +48,7 @@ typedef struct Oid transientoid; /* OID of new heap into which to store */ /* These fields are filled by transientrel_startup: */ Relation transientrel; /* relation to write to */ - CommandId output_cid; /* cmin to insert in output tuples */ - int ti_options; /* table_tuple_insert performance options */ - BulkInsertState bistate; /* bulk insert state */ + TableModifyState *mstate; /* table insert state */ } DR_transientrel; static int matview_maintenance_depth = 0; @@ -458,9 +456,13 @@ transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo) * Fill private fields of myState for use by later routines */ myState->transientrel = transientrel; - myState->output_cid = GetCurrentCommandId(true); - myState->ti_options = TABLE_INSERT_SKIP_FSM | TABLE_INSERT_FROZEN; - myState->bistate = GetBulkInsertState(); + myState->mstate = table_modify_begin(transientrel, + TM_FLAG_BAS_BULKWRITE, + GetCurrentCommandId(true), + TABLE_INSERT_SKIP_FSM | + TABLE_INSERT_FROZEN, + NULL, + NULL); /* * Valid smgr_targblock implies something already wrote to the relation. @@ -485,12 +487,7 @@ transientrel_receive(TupleTableSlot *slot, DestReceiver *self) * cheap either. This also doesn't allow accessing per-AM data (say a * tuple's xmin), but since we don't do that here... */ - - table_tuple_insert(myState->transientrel, - slot, - myState->output_cid, - myState->ti_options, - myState->bistate); + table_modify_buffer_insert(myState->mstate, slot); /* We know this is a newly created relation, so there are no indexes */ @@ -505,9 +502,7 @@ transientrel_shutdown(DestReceiver *self) { DR_transientrel *myState = (DR_transientrel *) self; - FreeBulkInsertState(myState->bistate); - - table_finish_bulk_insert(myState->transientrel, myState->ti_options); + table_modify_end(myState->mstate); /* close transientrel, but keep lock until commit */ table_close(myState->transientrel, NoLock); diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 7b6c69b7a5..9b79f46d7a 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -5966,10 +5966,8 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode) int i; ListCell *l; EState *estate; - CommandId mycid; - BulkInsertState bistate; - int ti_options; ExprState *partqualstate = NULL; + TableModifyState *mstate = NULL; /* * Open the relation(s). We have surely already locked the existing @@ -5988,18 +5986,14 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode) * Prepare a BulkInsertState and options for table_tuple_insert. The FSM * is empty, so don't bother using it. */ - if (newrel) + if (newrel && mstate == NULL) { - mycid = GetCurrentCommandId(true); - bistate = GetBulkInsertState(); - ti_options = TABLE_INSERT_SKIP_FSM; - } - else - { - /* keep compiler quiet about using these uninitialized */ - mycid = 0; - bistate = NULL; - ti_options = 0; + mstate = table_modify_begin(newrel, + TM_FLAG_BAS_BULKWRITE, + GetCurrentCommandId(true), + TABLE_INSERT_SKIP_FSM, + NULL, + NULL); } /* @@ -6297,8 +6291,7 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode) /* Write the tuple out to the new relation */ if (newrel) - table_tuple_insert(newrel, insertslot, mycid, - ti_options, bistate); + table_modify_buffer_insert(mstate, insertslot); ResetExprContext(econtext); @@ -6319,10 +6312,7 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode) table_close(oldrel, NoLock); if (newrel) { - FreeBulkInsertState(bistate); - - table_finish_bulk_insert(newrel, ti_options); - + table_modify_end(mstate); table_close(newrel, NoLock); } } -- 2.34.1
From d30d0c31ceecac5a0f7d61b7e096225dcfdba3c9 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com> Date: Tue, 4 Jun 2024 20:23:03 +0000 Subject: [PATCH v22 3/5] Optimize INSERT INTO SELECT with new multi insert table AM This commit optimizes the INSERT INTO SELECT query for heap AM using new multi insert table AM added by commit <<CHANGE_ME>>. Author: Bharath Rupireddy Reviewed-by: Jeff Davis Discussion: https://www.postgresql.org/message-id/CALj2ACVi9eTRYR%3Dgdca5wxtj3Kk_9q9qVccxsS1hngTGOCjPwQ%40mail.gmail.com Discussion: https://www.postgresql.org/message-id/8633171cb034aafc260fdf37df04b6c779aa1e2f.camel%40j-davis.com --- src/backend/executor/nodeModifyTable.c | 170 ++++++++++++++++++++++--- src/tools/pgindent/typedefs.list | 1 + 2 files changed, 153 insertions(+), 18 deletions(-) diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index cee60d3659..582c4a9842 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -114,6 +114,18 @@ typedef struct UpdateContext LockTupleMode lockmode; } UpdateContext; +typedef struct InsertModifyBufferFlushContext +{ + ResultRelInfo *resultRelInfo; + EState *estate; + ModifyTableState *mtstate; +} InsertModifyBufferFlushContext; + +static InsertModifyBufferFlushContext *insert_modify_buffer_flush_context = NULL; +static TableModifyState *table_modify_state = NULL; + +static void InsertModifyBufferFlushCallback(void *context, + TupleTableSlot *slot); static void ExecBatchInsert(ModifyTableState *mtstate, ResultRelInfo *resultRelInfo, @@ -726,6 +738,55 @@ ExecGetUpdateNewTuple(ResultRelInfo *relinfo, return ExecProject(newProj); } +static void +InsertModifyBufferFlushCallback(void *context, TupleTableSlot *slot) +{ + InsertModifyBufferFlushContext *ctx = (InsertModifyBufferFlushContext *) context; + ResultRelInfo *resultRelInfo = ctx->resultRelInfo; + EState *estate = ctx->estate; + ModifyTableState *mtstate = ctx->mtstate; + + /* Quick exit if no indexes or no triggers */ + if (!(resultRelInfo->ri_NumIndices > 0 || + (resultRelInfo->ri_TrigDesc != NULL && + (resultRelInfo->ri_TrigDesc->trig_insert_after_row || + resultRelInfo->ri_TrigDesc->trig_insert_new_table)))) + return; + + /* Caller must take care of opening and closing the indices */ + + /* + * 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; + + recheckIndexes = + ExecInsertIndexTuples(resultRelInfo, + slot, estate, false, + false, NULL, NIL, false); + ExecARInsertTriggers(estate, resultRelInfo, + slot, recheckIndexes, + mtstate->mt_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)) + { + ExecARInsertTriggers(estate, resultRelInfo, + slot, NIL, + mtstate->mt_transition_capture); + } +} + /* ---------------------------------------------------------------- * ExecInsert * @@ -751,7 +812,8 @@ ExecInsert(ModifyTableContext *context, TupleTableSlot *slot, bool canSetTag, TupleTableSlot **inserted_tuple, - ResultRelInfo **insert_destrel) + ResultRelInfo **insert_destrel, + bool canMultiInsert) { ModifyTableState *mtstate = context->mtstate; EState *estate = context->estate; @@ -764,6 +826,7 @@ ExecInsert(ModifyTableContext *context, OnConflictAction onconflict = node->onConflictAction; PartitionTupleRouting *proute = mtstate->mt_partition_tuple_routing; MemoryContext oldContext; + bool ar_insert_triggers_executed = false; /* * If the input result relation is a partitioned table, find the leaf @@ -1126,17 +1189,53 @@ ExecInsert(ModifyTableContext *context, } else { - /* insert the tuple normally */ - table_tuple_insert(resultRelationDesc, slot, - estate->es_output_cid, - 0, NULL); + if (canMultiInsert && + proute == NULL && + resultRelInfo->ri_WithCheckOptions == NIL && + resultRelInfo->ri_projectReturning == NULL) + { + if (insert_modify_buffer_flush_context == NULL) + { + insert_modify_buffer_flush_context = + (InsertModifyBufferFlushContext *) palloc0(sizeof(InsertModifyBufferFlushContext)); + insert_modify_buffer_flush_context->resultRelInfo = resultRelInfo; + insert_modify_buffer_flush_context->estate = estate; + insert_modify_buffer_flush_context->mtstate = mtstate; + } - /* insert index entries for tuple */ - if (resultRelInfo->ri_NumIndices > 0) - recheckIndexes = ExecInsertIndexTuples(resultRelInfo, - slot, estate, false, - false, NULL, NIL, - false); + if (table_modify_state == NULL) + { + table_modify_state = table_modify_begin(resultRelInfo->ri_RelationDesc, + 0, + estate->es_output_cid, + 0, + InsertModifyBufferFlushCallback, + insert_modify_buffer_flush_context); + } + + table_modify_buffer_insert(table_modify_state, slot); + ar_insert_triggers_executed = true; + } + else + { + /* insert the tuple normally */ + table_tuple_insert(resultRelationDesc, slot, + estate->es_output_cid, + 0, NULL); + + /* insert index entries for tuple */ + if (resultRelInfo->ri_NumIndices > 0) + recheckIndexes = ExecInsertIndexTuples(resultRelInfo, + slot, estate, false, + false, NULL, NIL, + false); + + ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes, + mtstate->mt_transition_capture); + + list_free(recheckIndexes); + ar_insert_triggers_executed = true; + } } } @@ -1170,10 +1269,12 @@ ExecInsert(ModifyTableContext *context, } /* AFTER ROW INSERT Triggers */ - ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes, - ar_insert_trig_tcs); - - list_free(recheckIndexes); + if (!ar_insert_triggers_executed) + { + ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes, + ar_insert_trig_tcs); + list_free(recheckIndexes); + } /* * Check any WITH CHECK OPTION constraints from parent views. We are @@ -1869,7 +1970,7 @@ ExecCrossPartitionUpdate(ModifyTableContext *context, /* Tuple routing starts from the root table. */ context->cpUpdateReturningSlot = ExecInsert(context, mtstate->rootResultRelInfo, slot, canSetTag, - inserted_tuple, insert_destrel); + inserted_tuple, insert_destrel, false); /* * Reset the transition state that may possibly have been written by @@ -3364,7 +3465,7 @@ ExecMergeNotMatched(ModifyTableContext *context, ResultRelInfo *resultRelInfo, mtstate->mt_merge_action = action; rslot = ExecInsert(context, mtstate->rootResultRelInfo, - newslot, canSetTag, NULL, NULL); + newslot, canSetTag, NULL, NULL, false); mtstate->mt_merge_inserted += 1; break; case CMD_NOTHING: @@ -3749,6 +3850,10 @@ ExecModifyTable(PlanState *pstate) HeapTupleData oldtupdata; HeapTuple oldtuple; ItemPointer tupleid; + bool canMultiInsert = false; + + table_modify_state = NULL; + insert_modify_buffer_flush_context = NULL; CHECK_FOR_INTERRUPTS(); @@ -3844,6 +3949,10 @@ ExecModifyTable(PlanState *pstate) if (TupIsNull(context.planSlot)) break; + if (operation == CMD_INSERT && + nodeTag(subplanstate) == T_SeqScanState) + canMultiInsert = true; + /* * When there are multiple result relations, each tuple contains a * junk column that gives the OID of the rel from which it came. @@ -4057,7 +4166,7 @@ ExecModifyTable(PlanState *pstate) ExecInitInsertProjection(node, resultRelInfo); slot = ExecGetInsertNewTuple(resultRelInfo, context.planSlot); slot = ExecInsert(&context, resultRelInfo, slot, - node->canSetTag, NULL, NULL); + node->canSetTag, NULL, NULL, canMultiInsert); break; case CMD_UPDATE: @@ -4116,6 +4225,17 @@ ExecModifyTable(PlanState *pstate) return slot; } + if (table_modify_state != NULL) + { + Assert(operation == CMD_INSERT); + + table_modify_end(table_modify_state); + table_modify_state = NULL; + + pfree(insert_modify_buffer_flush_context); + insert_modify_buffer_flush_context = NULL; + } + /* * Insert remaining tuples for batch insert. */ @@ -4228,6 +4348,9 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) mtstate->mt_merge_updated = 0; mtstate->mt_merge_deleted = 0; + table_modify_state = NULL; + insert_modify_buffer_flush_context = NULL; + /*---------- * Resolve the target relation. This is the same as: * @@ -4681,6 +4804,17 @@ ExecEndModifyTable(ModifyTableState *node) { int i; + if (table_modify_state != NULL) + { + Assert(node->operation == CMD_INSERT); + + table_modify_end(table_modify_state); + table_modify_state = NULL; + + pfree(insert_modify_buffer_flush_context); + insert_modify_buffer_flush_context = NULL; + } + /* * Allow any FDWs to shut down */ diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 84baf9b78d..3353faa6cd 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1180,6 +1180,7 @@ ImportForeignSchema_function ImportQual InProgressEnt InProgressIO +InsertModifyBufferFlushContext IncludeWal InclusionOpaque IncrementVarSublevelsUp_context -- 2.34.1
From c7c2e8ebff060cbd36e508c7a0c28ba6696c427d Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com> Date: Tue, 4 Jun 2024 20:24:21 +0000 Subject: [PATCH v22 4/5] Optimize Logical Replication Apply with new multi insert table AM This commit optimizes the Logical Replication Apply for heap AM using new multi insert table AM added by commit <<CHANGE_ME>>. Author: Bharath Rupireddy Reviewed-by: Jeff Davis Discussion: https://www.postgresql.org/message-id/CALj2ACVi9eTRYR%3Dgdca5wxtj3Kk_9q9qVccxsS1hngTGOCjPwQ%40mail.gmail.com Discussion: https://www.postgresql.org/message-id/8633171cb034aafc260fdf37df04b6c779aa1e2f.camel%40j-davis.com --- src/backend/executor/execReplication.c | 39 +++ src/backend/replication/logical/proto.c | 24 ++ src/backend/replication/logical/worker.c | 350 ++++++++++++++++++++++- src/include/executor/executor.h | 4 + src/include/replication/logicalproto.h | 2 + src/tools/pgindent/typedefs.list | 2 + 6 files changed, 408 insertions(+), 13 deletions(-) diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index d0a89cd577..fae1375537 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -544,6 +544,45 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, } } +void +ExecRelationMultiInsert(TableModifyState *MultiInsertState, + ResultRelInfo *resultRelInfo, + EState *estate, TupleTableSlot *slot) +{ + bool skip_tuple = false; + Relation rel = resultRelInfo->ri_RelationDesc; + + /* For now we support only tables. */ + Assert(rel->rd_rel->relkind == RELKIND_RELATION); + + CheckCmdReplicaIdentity(rel, CMD_INSERT); + + /* BEFORE ROW INSERT Triggers */ + if (resultRelInfo->ri_TrigDesc && + resultRelInfo->ri_TrigDesc->trig_insert_before_row) + { + if (!ExecBRInsertTriggers(estate, resultRelInfo, slot)) + skip_tuple = true; /* "do nothing" */ + } + + if (!skip_tuple) + { + /* Compute stored generated columns */ + if (rel->rd_att->constr && + rel->rd_att->constr->has_generated_stored) + ExecComputeStoredGenerated(resultRelInfo, estate, slot, + CMD_INSERT); + + /* Check the constraints of the tuple */ + if (rel->rd_att->constr) + ExecConstraints(resultRelInfo, slot, estate); + if (rel->rd_rel->relispartition) + ExecPartitionCheck(resultRelInfo, slot, estate, true); + + table_modify_buffer_insert(MultiInsertState, slot); + } +} + /* * Find the searchslot tuple and update it with data in the slot, * update the indexes, and execute any constraints and per-row triggers. diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 95c09c9516..46d38aebd2 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -427,6 +427,30 @@ logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel, logicalrep_write_tuple(out, rel, newslot, binary, columns); } +LogicalRepRelId +logicalrep_read_relid(StringInfo in) +{ + LogicalRepRelId relid; + + /* read the relation id */ + relid = pq_getmsgint(in, 4); + + return relid; +} + +void +logicalrep_read_insert_v2(StringInfo in, LogicalRepTupleData *newtup) +{ + char action; + + action = pq_getmsgbyte(in); + if (action != 'N') + elog(ERROR, "expected new tuple but got %d", + action); + + logicalrep_read_tuple(in, newtup); +} + /* * Read INSERT from stream. * diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index b5a80fe3e8..6d80f650ce 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -148,7 +148,6 @@ #include <unistd.h> #include "access/table.h" -#include "access/tableam.h" #include "access/twophase.h" #include "access/xact.h" #include "catalog/indexing.h" @@ -416,6 +415,30 @@ static inline void reset_apply_error_context_info(void); static TransApplyAction get_transaction_apply_action(TransactionId xid, ParallelApplyWorkerInfo **winfo); +typedef enum LRMultiInsertReturnStatus +{ + LR_MULTI_INSERT_NONE = 0, + LR_MULTI_INSERT_REL_SKIPPED, + LR_MULTI_INSERT_DISALLOWED, + LR_MULTI_INSERT_DONE, +} LRMultiInsertReturnStatus; + +static TableModifyState *MultiInsertState = NULL; +static LogicalRepRelMapEntry *LastRel = NULL; +static LogicalRepRelId LastMultiInsertRelId = InvalidOid; +static ApplyExecutionData *LastEData = NULL; +static TupleTableSlot *LastRemoteSlot = NULL; + +typedef struct LRModifyBufferFlushContext +{ + ResultRelInfo *resultRelInfo; + EState *estate; +} LRModifyBufferFlushContext; + +static LRModifyBufferFlushContext *modify_buffer_flush_context = NULL; +static void LRModifyBufferFlushCallback(void *context, TupleTableSlot *slot); +static void FinishMultiInserts(void); + /* * Form the origin name for the subscription. * @@ -1017,6 +1040,8 @@ apply_handle_commit(StringInfo s) { LogicalRepCommitData commit_data; + FinishMultiInserts(); + logicalrep_read_commit(s, &commit_data); if (commit_data.commit_lsn != remote_final_lsn) @@ -1043,6 +1068,8 @@ apply_handle_begin_prepare(StringInfo s) { LogicalRepPreparedTxnData begin_data; + FinishMultiInserts(); + /* Tablesync should never receive prepare. */ if (am_tablesync_worker()) ereport(ERROR, @@ -1109,6 +1136,8 @@ apply_handle_prepare(StringInfo s) { LogicalRepPreparedTxnData prepare_data; + FinishMultiInserts(); + logicalrep_read_prepare(s, &prepare_data); if (prepare_data.prepare_lsn != remote_final_lsn) @@ -1171,6 +1200,8 @@ apply_handle_commit_prepared(StringInfo s) LogicalRepCommitPreparedTxnData prepare_data; char gid[GIDSIZE]; + FinishMultiInserts(); + logicalrep_read_commit_prepared(s, &prepare_data); set_apply_error_context_xact(prepare_data.xid, prepare_data.commit_lsn); @@ -1220,6 +1251,8 @@ apply_handle_rollback_prepared(StringInfo s) LogicalRepRollbackPreparedTxnData rollback_data; char gid[GIDSIZE]; + FinishMultiInserts(); + logicalrep_read_rollback_prepared(s, &rollback_data); set_apply_error_context_xact(rollback_data.xid, rollback_data.rollback_end_lsn); @@ -1277,6 +1310,8 @@ apply_handle_stream_prepare(StringInfo s) /* Save the message before it is consumed. */ StringInfoData original_msg = *s; + FinishMultiInserts(); + if (in_streamed_transaction) ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), @@ -1304,6 +1339,8 @@ apply_handle_stream_prepare(StringInfo s) apply_spooled_messages(MyLogicalRepWorker->stream_fileset, prepare_data.xid, prepare_data.prepare_lsn); + FinishMultiInserts(); + /* Mark the transaction as prepared. */ apply_handle_prepare_internal(&prepare_data); @@ -1407,6 +1444,8 @@ apply_handle_stream_prepare(StringInfo s) static void apply_handle_origin(StringInfo s) { + FinishMultiInserts(); + /* * ORIGIN message can only come inside streaming transaction or inside * remote transaction and before any actual writes. @@ -1473,6 +1512,8 @@ apply_handle_stream_start(StringInfo s) /* Save the message before it is consumed. */ StringInfoData original_msg = *s; + FinishMultiInserts(); + if (in_streamed_transaction) ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), @@ -1628,6 +1669,8 @@ apply_handle_stream_stop(StringInfo s) ParallelApplyWorkerInfo *winfo; TransApplyAction apply_action; + FinishMultiInserts(); + if (!in_streamed_transaction) ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), @@ -1821,6 +1864,8 @@ apply_handle_stream_abort(StringInfo s) StringInfoData original_msg = *s; bool toplevel_xact; + FinishMultiInserts(); + if (in_streamed_transaction) ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), @@ -2138,6 +2183,8 @@ apply_handle_stream_commit(StringInfo s) /* Save the message before it is consumed. */ StringInfoData original_msg = *s; + FinishMultiInserts(); + if (in_streamed_transaction) ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), @@ -2159,6 +2206,8 @@ apply_handle_stream_commit(StringInfo s) apply_spooled_messages(MyLogicalRepWorker->stream_fileset, xid, commit_data.commit_lsn); + FinishMultiInserts(); + apply_handle_commit_internal(&commit_data); /* Unlink the files with serialized changes and subxact info. */ @@ -2302,6 +2351,8 @@ apply_handle_relation(StringInfo s) { LogicalRepRelation *rel; + FinishMultiInserts(); + if (handle_streamed_transaction(LOGICAL_REP_MSG_RELATION, s)) return; @@ -2325,6 +2376,8 @@ apply_handle_type(StringInfo s) { LogicalRepTyp typ; + FinishMultiInserts(); + if (handle_streamed_transaction(LOGICAL_REP_MSG_TYPE, s)) return; @@ -2363,16 +2416,126 @@ TargetPrivilegesCheck(Relation rel, AclMode mode) RelationGetRelationName(rel)))); } -/* - * Handle INSERT message. - */ +static void +FinishMultiInserts(void) +{ + LogicalRepMsgType saved_command; + + if (MultiInsertState == NULL) + return; + + Assert(OidIsValid(LastMultiInsertRelId)); + Assert(LastEData != NULL); + + /* Set relation for error callback */ + apply_error_callback_arg.rel = LastRel; + + /* Set current command for error callback */ + saved_command = apply_error_callback_arg.command; + apply_error_callback_arg.command = LOGICAL_REP_MSG_INSERT; + + ExecDropSingleTupleTableSlot(LastRemoteSlot); + LastRemoteSlot = NULL; + + table_modify_end(MultiInsertState); + MultiInsertState = NULL; + LastMultiInsertRelId = InvalidOid; + + pfree(modify_buffer_flush_context); + modify_buffer_flush_context = NULL; + + ExecCloseIndices(LastEData->targetRelInfo); + + finish_edata(LastEData); + LastEData = NULL; + + /* Reset relation for error callback */ + apply_error_callback_arg.rel = NULL; + + /* Reset the current command */ + apply_error_callback_arg.command = saved_command; + + logicalrep_rel_close(LastRel, NoLock); + LastRel = NULL; + + end_replication_step(); +} static void -apply_handle_insert(StringInfo s) +LRModifyBufferFlushCallback(void *context, TupleTableSlot *slot) +{ + LRModifyBufferFlushContext *ctx = (LRModifyBufferFlushContext *) context; + ResultRelInfo *resultRelInfo = ctx->resultRelInfo; + EState *estate = ctx->estate; + LogicalRepMsgType saved_command; + + /* Quick exit if no indexes or no triggers */ + if (!(resultRelInfo->ri_NumIndices > 0 || + (resultRelInfo->ri_TrigDesc != NULL && + (resultRelInfo->ri_TrigDesc->trig_insert_after_row || + resultRelInfo->ri_TrigDesc->trig_insert_new_table)))) + return; + + /* Set relation for error callback */ + apply_error_callback_arg.rel = LastRel; + + /* Set current command for error callback */ + saved_command = apply_error_callback_arg.command; + apply_error_callback_arg.command = LOGICAL_REP_MSG_INSERT; + + /* Caller must take care of opening and closing the indices */ + + /* + * 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; + + recheckIndexes = + ExecInsertIndexTuples(resultRelInfo, + slot, estate, false, + false, NULL, NIL, false); + + ExecARInsertTriggers(estate, resultRelInfo, + slot, recheckIndexes, + NULL); + + 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)) + { + ExecARInsertTriggers(estate, resultRelInfo, + slot, NIL, + NULL); + } + + /* + * XXX we should in theory pass a TransitionCaptureState object to the + * above to capture transition tuples, but after statement triggers don't + * actually get fired by replication yet anyway + */ + + /* Reset relation for error callback */ + apply_error_callback_arg.rel = NULL; + + /* Reset the current command */ + apply_error_callback_arg.command = saved_command; +} + +static LRMultiInsertReturnStatus +do_multi_inserts(StringInfo s, LogicalRepRelId *relid) { LogicalRepRelMapEntry *rel; LogicalRepTupleData newtup; - LogicalRepRelId relid; UserContext ucxt; ApplyExecutionData *edata; EState *estate; @@ -2380,17 +2543,142 @@ apply_handle_insert(StringInfo s) MemoryContext oldctx; bool run_as_owner; + if (MultiInsertState == NULL) + begin_replication_step(); + + *relid = logicalrep_read_relid(s); + + if (MultiInsertState != NULL && + (LastMultiInsertRelId != InvalidOid && + *relid != InvalidOid && + LastMultiInsertRelId != *relid)) + FinishMultiInserts(); + + if (MultiInsertState == NULL) + rel = logicalrep_rel_open(*relid, RowExclusiveLock); + else + rel = LastRel; + + if (!should_apply_changes_for_rel(rel)) + { + Assert(MultiInsertState == NULL); + + /* + * The relation can't become interesting in the middle of the + * transaction so it's safe to unlock it. + */ + logicalrep_rel_close(rel, RowExclusiveLock); + end_replication_step(); + return LR_MULTI_INSERT_REL_SKIPPED; + } + + /* For a partitioned table, let's not do multi inserts. */ + if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + { + Assert(MultiInsertState == NULL); + + /* + * The relation can't become interesting in the middle of the + * transaction so it's safe to unlock it. + */ + logicalrep_rel_close(rel, RowExclusiveLock); + end_replication_step(); + return LR_MULTI_INSERT_DISALLOWED; + } + /* - * Quick return if we are skipping data modification changes or handling - * streamed transactions. + * Make sure that any user-supplied code runs as the table owner, unless + * the user has opted out of that behavior. */ - if (is_skipping_changes() || - handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, s)) - return; + run_as_owner = MySubscription->runasowner; + if (!run_as_owner) + SwitchToUntrustedUser(rel->localrel->rd_rel->relowner, &ucxt); + + /* Set relation for error callback */ + apply_error_callback_arg.rel = rel; + + if (MultiInsertState == NULL) + { + oldctx = MemoryContextSwitchTo(TopTransactionContext); + + /* Initialize the executor state. */ + LastEData = edata = create_edata_for_relation(rel); + estate = edata->estate; + + LastRemoteSlot = remoteslot = MakeTupleTableSlot(RelationGetDescr(rel->localrel), + &TTSOpsVirtual); + + modify_buffer_flush_context = (LRModifyBufferFlushContext *) palloc(sizeof(LRModifyBufferFlushContext)); + modify_buffer_flush_context->resultRelInfo = edata->targetRelInfo; + modify_buffer_flush_context->estate = estate; + + MultiInsertState = table_modify_begin(edata->targetRelInfo->ri_RelationDesc, + TM_FLAG_BAS_BULKWRITE, + GetCurrentCommandId(true), + 0, + LRModifyBufferFlushCallback, + modify_buffer_flush_context); + LastRel = rel; + LastMultiInsertRelId = *relid; + + /* We must open indexes here. */ + ExecOpenIndices(edata->targetRelInfo, false); + + MemoryContextSwitchTo(oldctx); + } + else + { + CommandId cid; + + edata = LastEData; + estate = edata->estate; + ResetExprContext(GetPerTupleExprContext(estate)); + ExecClearTuple(LastRemoteSlot); + remoteslot = LastRemoteSlot; + cid = GetCurrentCommandId(true); + MultiInsertState->cid = cid; + estate->es_output_cid = cid; + } + + /* Process and store remote tuple in the slot */ + logicalrep_read_insert_v2(s, &newtup); + oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); + slot_store_data(remoteslot, rel, &newtup); + slot_fill_defaults(rel, estate, remoteslot); + MemoryContextSwitchTo(oldctx); + + TargetPrivilegesCheck(edata->targetRelInfo->ri_RelationDesc, ACL_INSERT); + ExecRelationMultiInsert(MultiInsertState, edata->targetRelInfo, estate, remoteslot); + + /* Reset relation for error callback */ + apply_error_callback_arg.rel = NULL; + + if (!run_as_owner) + RestoreUserContext(&ucxt); + + Assert(MultiInsertState != NULL); + + CommandCounterIncrement(); + + return LR_MULTI_INSERT_DONE; +} + +static bool +do_single_inserts(StringInfo s, LogicalRepRelId relid) +{ + LogicalRepRelMapEntry *rel; + LogicalRepTupleData newtup; + UserContext ucxt; + ApplyExecutionData *edata; + EState *estate; + TupleTableSlot *remoteslot; + MemoryContext oldctx; + bool run_as_owner; + + Assert(relid != InvalidOid); begin_replication_step(); - relid = logicalrep_read_insert(s, &newtup); rel = logicalrep_rel_open(relid, RowExclusiveLock); if (!should_apply_changes_for_rel(rel)) { @@ -2400,7 +2688,7 @@ apply_handle_insert(StringInfo s) */ logicalrep_rel_close(rel, RowExclusiveLock); end_replication_step(); - return; + return false; } /* @@ -2422,6 +2710,7 @@ apply_handle_insert(StringInfo s) &TTSOpsVirtual); /* Process and store remote tuple in the slot */ + logicalrep_read_insert_v2(s, &newtup); oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); slot_store_data(remoteslot, rel, &newtup); slot_fill_defaults(rel, estate, remoteslot); @@ -2446,6 +2735,35 @@ apply_handle_insert(StringInfo s) logicalrep_rel_close(rel, NoLock); end_replication_step(); + + return true; +} + +/* + * Handle INSERT message. + */ +static void +apply_handle_insert(StringInfo s) +{ + LRMultiInsertReturnStatus mi_status; + LogicalRepRelId relid; + + /* + * Quick return if we are skipping data modification changes or handling + * streamed transactions. + */ + if (is_skipping_changes() || + handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, s)) + return; + + mi_status = do_multi_inserts(s, &relid); + if (mi_status == LR_MULTI_INSERT_REL_SKIPPED || + mi_status == LR_MULTI_INSERT_DONE) + return; + + do_single_inserts(s, relid); + + return; } /* @@ -2532,6 +2850,8 @@ apply_handle_update(StringInfo s) MemoryContext oldctx; bool run_as_owner; + FinishMultiInserts(); + /* * Quick return if we are skipping data modification changes or handling * streamed transactions. @@ -2713,6 +3033,8 @@ apply_handle_delete(StringInfo s) MemoryContext oldctx; bool run_as_owner; + FinishMultiInserts(); + /* * Quick return if we are skipping data modification changes or handling * streamed transactions. @@ -3154,6 +3476,8 @@ apply_handle_truncate(StringInfo s) ListCell *lc; LOCKMODE lockmode = AccessExclusiveLock; + FinishMultiInserts(); + /* * Quick return if we are skipping data modification changes or handling * streamed transactions. diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index 9770752ea3..8f10ea977b 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -14,6 +14,7 @@ #ifndef EXECUTOR_H #define EXECUTOR_H +#include "access/tableam.h" #include "executor/execdesc.h" #include "fmgr.h" #include "nodes/lockoptions.h" @@ -656,6 +657,9 @@ extern bool RelationFindReplTupleSeq(Relation rel, LockTupleMode lockmode, extern void ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, EState *estate, TupleTableSlot *slot); +extern void ExecRelationMultiInsert(TableModifyState *MultiInsertState, + ResultRelInfo *resultRelInfo, + EState *estate, TupleTableSlot *slot); extern void ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, TupleTableSlot *searchslot, TupleTableSlot *slot); diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index c409638a2e..3f3a7f0a31 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -226,6 +226,8 @@ extern void logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *newslot, bool binary, Bitmapset *columns); +extern LogicalRepRelId logicalrep_read_relid(StringInfo in); +extern void logicalrep_read_insert_v2(StringInfo in, LogicalRepTupleData *newtup); extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup); extern void logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 3353faa6cd..42f27dda70 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1470,6 +1470,8 @@ LPTHREAD_START_ROUTINE LPTSTR LPVOID LPWSTR +LRModifyBufferFlushContext +LRMultiInsertReturnStatus LSEG LUID LVRelState -- 2.34.1
From 46462539749b00695ed54a57be89eda8c278fbec Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com> Date: Tue, 4 Jun 2024 20:27:12 +0000 Subject: [PATCH v22 5/5] Use new multi insert table AM for COPY This commit uses the new multi insert table AM added by commit <<CHANGE_ME>> for COPY ... FROM command. Author: Bharath Rupireddy Reviewed-by: Jeff Davis Discussion: https://www.postgresql.org/message-id/CALj2ACVi9eTRYR%3Dgdca5wxtj3Kk_9q9qVccxsS1hngTGOCjPwQ%40mail.gmail.com Discussion: https://www.postgresql.org/message-id/8633171cb034aafc260fdf37df04b6c779aa1e2f.camel%40j-davis.com --- src/backend/commands/copyfrom.c | 234 +++++++++++++++-------- src/include/commands/copyfrom_internal.h | 4 +- src/tools/pgindent/typedefs.list | 1 + 3 files changed, 158 insertions(+), 81 deletions(-) diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index ce4d62e707..82b6f7faa1 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -71,14 +71,25 @@ /* Trim the list of buffers back down to this number after flushing */ #define MAX_PARTITION_BUFFERS 32 +typedef struct CopyModifyBufferFlushContext +{ + CopyFromState cstate; + ResultRelInfo *resultRelInfo; + EState *estate; +} CopyModifyBufferFlushContext; + /* Stores multi-insert data related to a single relation in CopyFrom. */ typedef struct CopyMultiInsertBuffer { - TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */ + TableModifyState *mstate; /* Table insert state; NULL if foreign table */ + TupleTableSlot **slots; /* Array to store tuples */ ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */ - BulkInsertState bistate; /* BulkInsertState for this rel if plain - * table; NULL if foreign table */ + TupleTableSlot *multislot; + CopyModifyBufferFlushContext *modify_buffer_flush_context; int nused; /* number of 'slots' containing tuples */ + int currslotno; /* Current buffered slot number that's being + * flushed; Used to get correct cur_lineno for + * errors while in flush callback. */ uint64 linenos[MAX_BUFFERED_TUPLES]; /* Line # of tuple in copy * stream */ } CopyMultiInsertBuffer; @@ -99,6 +110,7 @@ typedef struct CopyMultiInsertInfo int ti_options; /* table insert options */ } CopyMultiInsertInfo; +static void CopyModifyBufferFlushCallback(void *context, TupleTableSlot *slot); /* non-export function prototypes */ static void ClosePipeFromProgram(CopyFromState cstate); @@ -218,14 +230,37 @@ CopyLimitPrintoutLength(const char *str) * ResultRelInfo. */ static CopyMultiInsertBuffer * -CopyMultiInsertBufferInit(ResultRelInfo *rri) +CopyMultiInsertBufferInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, + CopyFromState cstate, EState *estate) { CopyMultiInsertBuffer *buffer; buffer = (CopyMultiInsertBuffer *) palloc(sizeof(CopyMultiInsertBuffer)); - memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES); + + if (rri->ri_FdwRoutine == NULL) + { + buffer->modify_buffer_flush_context = (CopyModifyBufferFlushContext *) palloc(sizeof(CopyModifyBufferFlushContext)); + buffer->modify_buffer_flush_context->cstate = cstate; + buffer->modify_buffer_flush_context->resultRelInfo = rri; + buffer->modify_buffer_flush_context->estate = estate; + + buffer->mstate = table_modify_begin(rri->ri_RelationDesc, + TM_FLAG_BAS_BULKWRITE, + miinfo->mycid, + miinfo->ti_options, + CopyModifyBufferFlushCallback, + buffer->modify_buffer_flush_context); + buffer->slots = NULL; + buffer->multislot = NULL; + } + else + { + buffer->mstate = NULL; + buffer->slots = palloc0(sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES); + buffer->multislot = NULL; + } + buffer->resultRelInfo = rri; - buffer->bistate = (rri->ri_FdwRoutine == NULL) ? GetBulkInsertState() : NULL; buffer->nused = 0; return buffer; @@ -236,11 +271,12 @@ CopyMultiInsertBufferInit(ResultRelInfo *rri) */ static inline void CopyMultiInsertInfoSetupBuffer(CopyMultiInsertInfo *miinfo, - ResultRelInfo *rri) + ResultRelInfo *rri, CopyFromState cstate, + EState *estate) { CopyMultiInsertBuffer *buffer; - buffer = CopyMultiInsertBufferInit(rri); + buffer = CopyMultiInsertBufferInit(miinfo, rri, cstate, estate); /* Setup back-link so we can easily find this buffer again */ rri->ri_CopyMultiInsertBuffer = buffer; @@ -273,7 +309,7 @@ CopyMultiInsertInfoInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, * tuples their way for the first time. */ if (rri->ri_RelationDesc->rd_rel->relkind != RELKIND_PARTITIONED_TABLE) - CopyMultiInsertInfoSetupBuffer(miinfo, rri); + CopyMultiInsertInfoSetupBuffer(miinfo, rri, cstate, estate); } /* @@ -317,8 +353,6 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, int batch_size = resultRelInfo->ri_BatchSize; int sent = 0; - Assert(buffer->bistate == NULL); - /* Ensure that the FDW supports batching and it's enabled */ Assert(resultRelInfo->ri_FdwRoutine->ExecForeignBatchInsert); Assert(batch_size > 1); @@ -390,13 +424,8 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, } else { - CommandId mycid = miinfo->mycid; - int ti_options = miinfo->ti_options; bool line_buf_valid = cstate->line_buf_valid; uint64 save_cur_lineno = cstate->cur_lineno; - MemoryContext oldcontext; - - Assert(buffer->bistate != NULL); /* * Print error context information correctly, if one of the operations @@ -404,56 +433,18 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, */ cstate->line_buf_valid = false; - /* - * table_multi_insert may leak memory, so switch to short-lived memory - * context before calling it. - */ - oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); - table_multi_insert(resultRelInfo->ri_RelationDesc, - slots, - nused, - mycid, - ti_options, - buffer->bistate); - MemoryContextSwitchTo(oldcontext); + Assert(buffer->currslotno <= buffer->nused); + buffer->currslotno = 0; - for (i = 0; i < nused; i++) - { - /* - * If there are any indexes, update them for all the inserted - * tuples, and run AFTER ROW INSERT triggers. - */ - if (resultRelInfo->ri_NumIndices > 0) - { - List *recheckIndexes; - - cstate->cur_lineno = buffer->linenos[i]; - recheckIndexes = - ExecInsertIndexTuples(resultRelInfo, - buffer->slots[i], estate, false, - false, NULL, NIL, false); - ExecARInsertTriggers(estate, resultRelInfo, - slots[i], recheckIndexes, - cstate->transition_capture); - list_free(recheckIndexes); - } + table_modify_buffer_flush(buffer->mstate); - /* - * 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); - } + Assert(buffer->currslotno <= buffer->nused); + buffer->currslotno = 0; - ExecClearTuple(slots[i]); - } + /* + * Indexes are updated and AFTER ROW INSERT triggers (if any) are run + * in the flush callback CopyModifyBufferFlushCallback. + */ /* Update the row counter and progress of the COPY command */ *processed += nused; @@ -469,6 +460,60 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, buffer->nused = 0; } +static void +CopyModifyBufferFlushCallback(void *context, TupleTableSlot *slot) +{ + CopyModifyBufferFlushContext *ctx = (CopyModifyBufferFlushContext *) context; + CopyFromState cstate = ctx->cstate; + ResultRelInfo *resultRelInfo = ctx->resultRelInfo; + EState *estate = ctx->estate; + CopyMultiInsertBuffer *buffer = resultRelInfo->ri_CopyMultiInsertBuffer; + + /* Quick exit if no indexes or no triggers */ + if (!(resultRelInfo->ri_NumIndices > 0 || + (resultRelInfo->ri_TrigDesc != NULL && + (resultRelInfo->ri_TrigDesc->trig_insert_after_row || + resultRelInfo->ri_TrigDesc->trig_insert_new_table)))) + return; + + /* Caller must take care of opening and closing the indices */ + + /* + * 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[buffer->currslotno++]; + recheckIndexes = + ExecInsertIndexTuples(resultRelInfo, + slot, estate, false, + false, NULL, NIL, false); + ExecARInsertTriggers(estate, resultRelInfo, + slot, 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[buffer->currslotno++]; + ExecARInsertTriggers(estate, resultRelInfo, + slot, NIL, + cstate->transition_capture); + } + + Assert(buffer->currslotno <= buffer->nused); +} + /* * Drop used slots and free member for this buffer. * @@ -489,19 +534,18 @@ CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo, if (resultRelInfo->ri_FdwRoutine == NULL) { - Assert(buffer->bistate != NULL); - FreeBulkInsertState(buffer->bistate); + table_modify_end(buffer->mstate); + ExecDropSingleTupleTableSlot(buffer->multislot); + pfree(buffer->modify_buffer_flush_context); } else - Assert(buffer->bistate == NULL); - - /* Since we only create slots on demand, just drop the non-null ones. */ - for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++) - ExecDropSingleTupleTableSlot(buffer->slots[i]); + { + /* Since we only create slots on demand, just drop the non-null ones. */ + for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++) + ExecDropSingleTupleTableSlot(buffer->slots[i]); - if (resultRelInfo->ri_FdwRoutine == NULL) - table_finish_bulk_insert(resultRelInfo->ri_RelationDesc, - miinfo->ti_options); + pfree(buffer->slots); + } pfree(buffer); } @@ -588,13 +632,32 @@ CopyMultiInsertInfoNextFreeSlot(CopyMultiInsertInfo *miinfo, { CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer; int nused = buffer->nused; + TupleTableSlot *slot; 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]; + if (rri->ri_FdwRoutine == NULL) + { + if (buffer->multislot == NULL) + buffer->multislot = MakeTupleTableSlot(RelationGetDescr(rri->ri_RelationDesc), + &TTSOpsVirtual); + + /* Caller must clear the slot */ + slot = buffer->multislot; + } + else + { + if (buffer->slots[nused] == NULL) + { + slot = table_slot_create(rri->ri_RelationDesc, NULL); + buffer->slots[nused] = slot; + } + else + slot = buffer->slots[nused]; + } + + return slot; } /* @@ -608,7 +671,11 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer; Assert(buffer != NULL); - Assert(slot == buffer->slots[buffer->nused]); + +#ifdef USE_ASSERT_CHECKING + if (rri->ri_FdwRoutine != NULL) + Assert(slot == buffer->slots[buffer->nused]); +#endif /* Store the line number so we can properly report any errors later */ buffer->linenos[buffer->nused] = lineno; @@ -616,6 +683,14 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, /* Record this slot as being used */ buffer->nused++; + if (rri->ri_FdwRoutine == NULL) + { + Assert(slot == buffer->multislot); + buffer->currslotno = 0; + + table_modify_buffer_insert(buffer->mstate, slot); + } + /* Update how many tuples are stored and their size */ miinfo->bufferedTuples++; miinfo->bufferedBytes += tuplen; @@ -830,7 +905,7 @@ CopyFrom(CopyFromState cstate) /* * It's generally more efficient to prepare a bunch of tuples for * insertion, and insert them in one - * table_multi_insert()/ExecForeignBatchInsert() call, than call + * table_modify_buffer_insert()/ExecForeignBatchInsert() call, than call * table_tuple_insert()/ExecForeignInsert() separately for every tuple. * However, there are a number of reasons why we might not be able to do * this. These are explained below. @@ -1080,7 +1155,8 @@ CopyFrom(CopyFromState cstate) { if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL) CopyMultiInsertInfoSetupBuffer(&multiInsertInfo, - resultRelInfo); + resultRelInfo, cstate, + estate); } else if (insertMethod == CIM_MULTI_CONDITIONAL && !CopyMultiInsertInfoIsEmpty(&multiInsertInfo)) diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h index cad52fcc78..14addbc6f6 100644 --- a/src/include/commands/copyfrom_internal.h +++ b/src/include/commands/copyfrom_internal.h @@ -46,9 +46,9 @@ typedef enum EolType typedef enum CopyInsertMethod { CIM_SINGLE, /* use table_tuple_insert or ExecForeignInsert */ - CIM_MULTI, /* always use table_multi_insert or + CIM_MULTI, /* always use table_modify_buffer_insert or * ExecForeignBatchInsert */ - CIM_MULTI_CONDITIONAL, /* use table_multi_insert or + CIM_MULTI_CONDITIONAL, /* use table_modify_buffer_insert or * ExecForeignBatchInsert only if valid */ } CopyInsertMethod; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 42f27dda70..4740c6946e 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -494,6 +494,7 @@ CopyHeaderChoice CopyInsertMethod CopyLogVerbosityChoice CopyMethod +CopyModifyBufferFlushContext CopyMultiInsertBuffer CopyMultiInsertInfo CopyOnErrorChoice -- 2.34.1
