From d69e7f0810d78b80a63a5dff65425daa5e0731c7 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Thu, 21 Mar 2024 07:02:43 +0000 Subject: [PATCH v13 1/4] New table AMs for single and multi inserts --- src/backend/access/heap/heapam.c | 235 +++++++++++++++++++++++ src/backend/access/heap/heapam_handler.c | 9 + src/include/access/heapam.h | 49 +++++ src/include/access/tableam.h | 138 +++++++++++++ 4 files changed, 431 insertions(+) diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 34bc60f625..fd2b3814dd 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" @@ -2442,6 +2443,240 @@ heap_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples, pgstat_count_heap_insert(relation, ntuples); } +/* + * Initialize state required for an insert a single tuple or multiple tuples + * into a heap. + */ +TableInsertState * +heap_insert_begin(Relation rel, CommandId cid, int am_flags, int insert_flags) +{ + TableInsertState *tistate; + + tistate = palloc0(sizeof(TableInsertState)); + tistate->rel = rel; + tistate->cid = cid; + tistate->am_flags = am_flags; + tistate->insert_flags = insert_flags; + + if ((am_flags & TABLEAM_MULTI_INSERTS) != 0 || + (am_flags & TABLEAM_BULKWRITE_BUFFER_ACCESS_STRATEGY)) + tistate->am_data = palloc0(sizeof(HeapInsertState)); + + if ((am_flags & TABLEAM_MULTI_INSERTS) != 0) + { + HeapMultiInsertState *mistate; + + mistate = palloc0(sizeof(HeapMultiInsertState)); + mistate->slots = palloc0(sizeof(TupleTableSlot *) * HEAP_MAX_BUFFERED_SLOTS); + + mistate->context = AllocSetContextCreate(CurrentMemoryContext, + "heap_multi_insert_v2 memory context", + ALLOCSET_DEFAULT_SIZES); + + ((HeapInsertState *) tistate->am_data)->mistate = mistate; + } + + if ((am_flags & TABLEAM_BULKWRITE_BUFFER_ACCESS_STRATEGY) != 0) + ((HeapInsertState *) tistate->am_data)->bistate = GetBulkInsertState(); + + return tistate; +} + +/* + * Insert a single tuple into a heap. + */ +void +heap_insert_v2(TableInsertState * state, TupleTableSlot *slot) +{ + bool shouldFree = true; + HeapTuple tuple = ExecFetchSlotHeapTuple(slot, true, &shouldFree); + BulkInsertState bistate = NULL; + + Assert(state->am_data != NULL && + ((HeapInsertState *) state->am_data)->mistate == NULL); + + /* Update tuple with table oid */ + slot->tts_tableOid = RelationGetRelid(state->rel); + tuple->t_tableOid = slot->tts_tableOid; + + if (state->am_data != NULL && + ((HeapInsertState *) state->am_data)->bistate != NULL) + bistate = ((HeapInsertState *) state->am_data)->bistate; + + /* Perform insertion, and copy the resulting ItemPointer */ + heap_insert(state->rel, tuple, state->cid, state->insert_flags, + bistate); + ItemPointerCopy(&tuple->t_self, &slot->tts_tid); + + if (shouldFree) + pfree(tuple); +} + +/* + * Create/return next free slot from multi-insert buffered slots array. + */ +TupleTableSlot * +heap_multi_insert_next_free_slot(TableInsertState * state) +{ + TupleTableSlot *slot; + HeapMultiInsertState *mistate; + + Assert(state->am_data != NULL && + ((HeapInsertState *) state->am_data)->mistate != NULL); + + mistate = ((HeapInsertState *) state->am_data)->mistate; + slot = mistate->slots[mistate->cur_slots]; + + if (slot == NULL) + { + slot = MakeTupleTableSlot(RelationGetDescr(state->rel), + &TTSOpsVirtual); + mistate->slots[mistate->cur_slots] = slot; + } + else + ExecClearTuple(slot); + + return slot; +} + +/* + * Store passed-in tuple into in-memory buffered slots. When full, insert + * multiple tuples from the buffers into heap. + */ +void +heap_multi_insert_v2(TableInsertState * state, TupleTableSlot *slot) +{ + TupleTableSlot *dstslot; + HeapMultiInsertState *mistate; + + Assert(state->am_data != NULL && + ((HeapInsertState *) state->am_data)->mistate != NULL); + + mistate = ((HeapInsertState *) state->am_data)->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; + } + + /* + * Caller may have got the slot using heap_multi_insert_next_free_slot, + * filled it and passed. So, skip copying in such a case. + */ + if ((state->am_flags & TABLEAM_SKIP_MULTI_INSERTS_FLUSH) == 0) + { + ExecClearTuple(dstslot); + ExecCopySlot(dstslot, slot); + } + else + Assert(dstslot == slot); + + mistate->cur_slots++; + + /* + * When passed-in slot is already materialized, memory allocated in slot's + * memory context is a close approximation for us to track the required + * space for the tuple in slot. + * + * For non-materialized slots, the flushing decision happens solely on the + * number of tuples stored in the buffer. + */ + if (TTS_SHOULDFREE(slot)) + mistate->cur_size += MemoryContextMemAllocated(slot->tts_mcxt, false); + + if ((state->am_flags & TABLEAM_SKIP_MULTI_INSERTS_FLUSH) == 0 && + (mistate->cur_slots >= HEAP_MAX_BUFFERED_SLOTS || + mistate->cur_size >= HEAP_MAX_BUFFERED_BYTES)) + heap_multi_insert_flush(state); +} + +/* + * Return pointer to multi-insert buffered slots array and number of currently + * occupied slots. + */ +TupleTableSlot ** +heap_multi_insert_slots(TableInsertState * state, int *num_slots) +{ + HeapMultiInsertState *mistate; + + mistate = ((HeapInsertState *) state->am_data)->mistate; + *num_slots = mistate->cur_slots; + + return mistate->slots; +} + +/* + * Insert multiple tuples from in-memory buffered slots into heap. + */ +void +heap_multi_insert_flush(TableInsertState * state) +{ + HeapMultiInsertState *mistate; + BulkInsertState bistate = NULL; + MemoryContext oldcontext; + + mistate = ((HeapInsertState *) state->am_data)->mistate; + + if (state->am_data != NULL && + ((HeapInsertState *) state->am_data)->bistate != NULL) + bistate = ((HeapInsertState *) state->am_data)->bistate; + + oldcontext = MemoryContextSwitchTo(mistate->context); + heap_multi_insert(state->rel, mistate->slots, mistate->cur_slots, + state->cid, state->insert_flags, bistate); + MemoryContextSwitchTo(oldcontext); + MemoryContextReset(mistate->context); + + mistate->cur_slots = 0; + mistate->cur_size = 0; +} + +/* + * Clean up state used to insert a single or multiple tuples into a heap. + */ +void +heap_insert_end(TableInsertState * state) +{ + if (state->am_data != NULL && + ((HeapInsertState *) state->am_data)->mistate != NULL) + { + HeapMultiInsertState *mistate = + ((HeapInsertState *) state->am_data)->mistate; + + /* Insert remaining tuples from multi-insert buffers */ + if (mistate->cur_slots > 0 || mistate->cur_size > 0) + heap_multi_insert_flush(state); + + MemoryContextDelete(mistate->context); + + for (int i = 0; i < HEAP_MAX_BUFFERED_SLOTS && mistate->slots[i] != NULL; i++) + ExecDropSingleTupleTableSlot(mistate->slots[i]); + + pfree(mistate); + ((HeapInsertState *) state->am_data)->mistate = NULL; + } + + if (state->am_data != NULL && + ((HeapInsertState *) state->am_data)->bistate != NULL) + FreeBulkInsertState(((HeapInsertState *) state->am_data)->bistate); + + pfree(state->am_data); + state->am_data = NULL; + pfree(state); +} + /* * 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 680a50bf8b..84793f324e 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -2562,6 +2562,15 @@ 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_insert_begin = heap_insert_begin, + .tuple_insert_v2 = heap_insert_v2, + .tuple_multi_insert_next_free_slot = heap_multi_insert_next_free_slot, + .tuple_multi_insert_v2 = heap_multi_insert_v2, + .tuple_multi_insert_slots = heap_multi_insert_slots, + .tuple_multi_insert_flush = heap_multi_insert_flush, + .tuple_insert_end = heap_insert_end, + .tuple_delete = heapam_tuple_delete, .tuple_update = heapam_tuple_update, .tuple_lock = heapam_tuple_lock, diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h index 4b133f6859..053be18110 100644 --- a/src/include/access/heapam.h +++ b/src/include/access/heapam.h @@ -225,6 +225,40 @@ htsv_get_valid_status(int status) return (HTSV_Result) status; } +/* + * 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. For instance, increasing this can cause + * quadratic growth in memory requirements during copies into partitioned + * tables with a large number of partitions. + */ +#define HEAP_MAX_BUFFERED_SLOTS 1000 + +/* Maximum size of all tuples that multi-insert buffers can hold */ +#define HEAP_MAX_BUFFERED_BYTES 65535 + +typedef struct HeapMultiInsertState +{ + /* Memory context to use for flushing multi-insert buffers */ + MemoryContext context; + + /* Array of buffered slots */ + TupleTableSlot **slots; + + /* Number of slots that multi-insert buffers currently hold */ + int cur_slots; + + /* Size of all tuples that multi-insert buffers currently hold */ + Size cur_size; +} HeapMultiInsertState; + +typedef struct HeapInsertState +{ + struct BulkInsertStateData *bistate; + HeapMultiInsertState *mistate; +} HeapInsertState; + /* ---------------- * function prototypes for heap access method * @@ -275,6 +309,21 @@ 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 TableInsertState * heap_insert_begin(Relation rel, + CommandId cid, + int am_flags, + int insert_flags); +extern void heap_insert_v2(TableInsertState * state, + TupleTableSlot *slot); +extern TupleTableSlot *heap_multi_insert_next_free_slot(TableInsertState * state); +extern void heap_multi_insert_v2(TableInsertState * state, + TupleTableSlot *slot); +extern TupleTableSlot **heap_multi_insert_slots(TableInsertState * state, + int *num_slots); +extern void heap_multi_insert_flush(TableInsertState * state); +extern void heap_insert_end(TableInsertState * 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 8249b37bbf..842cbdd16e 100644 --- a/src/include/access/tableam.h +++ b/src/include/access/tableam.h @@ -247,6 +247,43 @@ typedef struct TM_IndexDeleteOp TM_IndexStatus *status; } TM_IndexDeleteOp; +/* Use multi inserts, i.e. buffer multiple tuples and insert them at once */ +#define TABLEAM_MULTI_INSERTS 0x000001 + +/* Use BAS_BULKWRITE buffer access strategy */ +#define TABLEAM_BULKWRITE_BUFFER_ACCESS_STRATEGY 0x000002 + +/* + * Skip flushing buffered tuples automatically. Responsibility lies with the + * caller to flush the buffered tuples. + */ +#define TABLEAM_SKIP_MULTI_INSERTS_FLUSH 0x000004 + + +/* Holds table insert state. */ +typedef struct TableInsertState +{ + /* Table AM-agnostic data starts here */ + + Relation rel; /* Target relation */ + + /* + * Command ID for this insertion. If required, change this for each pass + * of insert functions. + */ + CommandId cid; + + /* Table AM options (TABLEAM_XXX macros) */ + int am_flags; + + /* table_tuple_insert performance options (TABLE_INSERT_XXX macros) */ + int insert_flags; + + /* Table AM specific data starts here */ + + void *am_data; +} TableInsertState; + /* "options" flag bits for table_tuple_insert */ /* TABLE_INSERT_SKIP_WAL was 0x0001; RelationNeedsWAL() now governs */ #define TABLE_INSERT_SKIP_FSM 0x0002 @@ -522,6 +559,20 @@ typedef struct TableAmRoutine void (*multi_insert) (Relation rel, TupleTableSlot **slots, int nslots, CommandId cid, int options, struct BulkInsertStateData *bistate); + TableInsertState *(*tuple_insert_begin) (Relation rel, + CommandId cid, + int am_flags, + int insert_flags); + void (*tuple_insert_v2) (TableInsertState * state, + TupleTableSlot *slot); + void (*tuple_multi_insert_v2) (TableInsertState * state, + TupleTableSlot *slot); + TupleTableSlot *(*tuple_multi_insert_next_free_slot) (TableInsertState * state); + TupleTableSlot **(*tuple_multi_insert_slots) (TableInsertState * state, + int *num_slots); + void (*tuple_multi_insert_flush) (TableInsertState * state); + void (*tuple_insert_end) (TableInsertState * state); + /* see table_tuple_delete() for reference about parameters */ TM_Result (*tuple_delete) (Relation rel, ItemPointer tid, @@ -1451,6 +1502,93 @@ table_multi_insert(Relation rel, TupleTableSlot **slots, int nslots, cid, options, bistate); } +static inline TableInsertState * +table_insert_begin(Relation rel, CommandId cid, int am_flags, + int insert_flags) +{ + if (rel->rd_tableam && rel->rd_tableam->tuple_insert_begin) + return rel->rd_tableam->tuple_insert_begin(rel, cid, am_flags, + insert_flags); + else + { + elog(ERROR, "table_insert_begin access method is not implemented for relation \"%s\"", + RelationGetRelationName(rel)); + return NULL; /* keep compiler quiet */ + } +} + +static inline void +table_tuple_insert_v2(TableInsertState * state, TupleTableSlot *slot) +{ + if (state->rel->rd_tableam && + state->rel->rd_tableam->tuple_insert_v2) + state->rel->rd_tableam->tuple_insert_v2(state, slot); + else + elog(ERROR, "table_tuple_insert_v2 access method is not implemented for relation \"%s\"", + RelationGetRelationName(state->rel)); +} + +static inline void +table_multi_insert_v2(TableInsertState * state, TupleTableSlot *slot) +{ + if (state->rel->rd_tableam && + state->rel->rd_tableam->tuple_multi_insert_v2) + state->rel->rd_tableam->tuple_multi_insert_v2(state, slot); + else + elog(ERROR, "table_multi_insert_v2 access method is not implemented for relation \"%s\"", + RelationGetRelationName(state->rel)); +} + +static inline TupleTableSlot * +table_multi_insert_next_free_slot(TableInsertState * state) +{ + if (state->rel->rd_tableam && + state->rel->rd_tableam->tuple_multi_insert_next_free_slot) + return state->rel->rd_tableam->tuple_multi_insert_next_free_slot(state); + else + { + elog(ERROR, "table_multi_insert_next_free_slot access method is not implemented for relation \"%s\"", + RelationGetRelationName(state->rel)); + return NULL; /* keep compiler quiet */ + } +} + +static inline TupleTableSlot ** +table_multi_insert_slots(TableInsertState * state, int *num_slots) +{ + if (state->rel->rd_tableam && + state->rel->rd_tableam->tuple_multi_insert_slots) + return state->rel->rd_tableam->tuple_multi_insert_slots(state, num_slots); + else + { + elog(ERROR, "table_multi_insert_slots access method is not implemented for relation \"%s\"", + RelationGetRelationName(state->rel)); + return NULL; /* keep compiler quiet */ + } +} + +static inline void +table_multi_insert_flush(TableInsertState * state) +{ + if (state->rel->rd_tableam && + state->rel->rd_tableam->tuple_multi_insert_flush) + state->rel->rd_tableam->tuple_multi_insert_flush(state); + else + elog(ERROR, "table_multi_insert_flush access method is not implemented for relation \"%s\"", + RelationGetRelationName(state->rel)); +} + +static inline void +table_insert_end(TableInsertState * state) +{ + if (state->rel->rd_tableam && + state->rel->rd_tableam->tuple_insert_end) + state->rel->rd_tableam->tuple_insert_end(state); + else + elog(ERROR, "table_insert_end access method is not implemented for relation \"%s\"", + RelationGetRelationName(state->rel)); +} + /* * Delete a tuple. * -- 2.34.1