diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index d687ceee33..3818efacb0 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -932,9 +932,11 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) tuplelen = datalen - SizeOfHeapHeader; change->data.tp.newtuple = - ReorderBufferGetTupleBuf(ctx->reorder, tuplelen); + ReorderBufferGetTupleBufForXid(ctx->reorder, XLogRecGetXid(r), + buf->origptr, tuplelen); - DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple); + if (change->data.tp.newtuple) + DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple); change->data.tp.clear_toast_afterwards = true; @@ -984,9 +986,13 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) tuplelen = datalen - SizeOfHeapHeader; change->data.tp.newtuple = - ReorderBufferGetTupleBuf(ctx->reorder, tuplelen); + ReorderBufferGetTupleBufForXid(ctx->reorder, + XLogRecGetXid(r), + buf->origptr, + tuplelen); - DecodeXLogTuple(data, datalen, change->data.tp.newtuple); + if (change->data.tp.newtuple) + DecodeXLogTuple(data, datalen, change->data.tp.newtuple); } if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD) @@ -1000,9 +1006,13 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) tuplelen = datalen - SizeOfHeapHeader; change->data.tp.oldtuple = - ReorderBufferGetTupleBuf(ctx->reorder, tuplelen); + ReorderBufferGetTupleBufForXid(ctx->reorder, + XLogRecGetXid(r), + buf->origptr, + tuplelen); - DecodeXLogTuple(data, datalen, change->data.tp.oldtuple); + if (change->data.tp.oldtuple) + DecodeXLogTuple(data, datalen, change->data.tp.oldtuple); } change->data.tp.clear_toast_afterwards = true; @@ -1055,10 +1065,14 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader)); change->data.tp.oldtuple = - ReorderBufferGetTupleBuf(ctx->reorder, tuplelen); - - DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete, - datalen, change->data.tp.oldtuple); + ReorderBufferGetTupleBufForXid(ctx->reorder, + XLogRecGetXid(r), + buf->origptr, + tuplelen); + + if (change->data.tp.oldtuple) + DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete, + datalen, change->data.tp.oldtuple); } change->data.tp.clear_toast_afterwards = true; @@ -1164,7 +1178,10 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) datalen = xlhdr->datalen; change->data.tp.newtuple = - ReorderBufferGetTupleBuf(ctx->reorder, datalen); + ReorderBufferGetTupleBufForXid(ctx->reorder, + XLogRecGetXid(r), + buf->origptr, + datalen); tuple = change->data.tp.newtuple; header = tuple->t_data; diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 00a8327e77..65b80fc156 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -336,17 +336,6 @@ ReorderBufferAllocate(void) SLAB_DEFAULT_BLOCK_SIZE, sizeof(ReorderBufferTXN)); - /* - * XXX the allocation sizes used below pre-date generation context's block - * growing code. These values should likely be benchmarked and set to - * more suitable values. - */ - buffer->tup_context = GenerationContextCreate(new_ctx, - "Tuples", - SLAB_LARGE_BLOCK_SIZE, - SLAB_LARGE_BLOCK_SIZE, - SLAB_LARGE_BLOCK_SIZE); - hash_ctl.keysize = sizeof(TransactionId); hash_ctl.entrysize = sizeof(ReorderBufferTXNByIdEnt); hash_ctl.hcxt = buffer->context; @@ -467,6 +456,10 @@ ReorderBufferReturnTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) /* Reset the toast hash */ ReorderBufferToastReset(rb, txn); + /* Remove the memory context for storing decoded tuples */ + if (txn->tup_context) + MemoryContextDelete(txn->tup_context); + pfree(txn); } @@ -504,17 +497,15 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change, case REORDER_BUFFER_CHANGE_UPDATE: case REORDER_BUFFER_CHANGE_DELETE: case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT: + /* + * Tuple buffers will be freed altogether when the tuple context + * is deleted. + */ if (change->data.tp.newtuple) - { - ReorderBufferReturnTupleBuf(change->data.tp.newtuple); change->data.tp.newtuple = NULL; - } - if (change->data.tp.oldtuple) - { - ReorderBufferReturnTupleBuf(change->data.tp.oldtuple); change->data.tp.oldtuple = NULL; - } + break; case REORDER_BUFFER_CHANGE_MESSAGE: if (change->data.msg.prefix != NULL) @@ -559,20 +550,57 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change, * overhead). */ HeapTuple -ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len) +ReorderBufferGetTupleBufForTXN(ReorderBuffer *rb, + ReorderBufferTXN *txn, + Size tuple_len) { HeapTuple tuple; Size alloc_len; - alloc_len = tuple_len + SizeofHeapTupleHeader; + if (txn->tup_context == NULL) + txn->tup_context = BumpContextCreate(rb->context, + "per-txn tuple context", + ALLOCSET_DEFAULT_SIZES); - tuple = (HeapTuple) MemoryContextAlloc(rb->tup_context, + alloc_len = tuple_len + SizeofHeapTupleHeader; + tuple = (HeapTuple) MemoryContextAlloc(txn->tup_context, HEAPTUPLESIZE + alloc_len); tuple->t_data = (HeapTupleHeader) ((char *) tuple + HEAPTUPLESIZE); return tuple; } +/* + * Similar to ReorderBufferGetTupleBufForTXN but for the transaction + * of the given xid. + */ +HeapTuple +ReorderBufferGetTupleBufForXid(ReorderBuffer *rb, TransactionId xid, + XLogRecPtr lsn, Size tuple_len) +{ + ReorderBufferTXN *txn; + + txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true); + Assert(txn != NULL); + + /* + * If the transaction is assigned to its top-level transaction, + * their decoded tuple data are stored to top-level transaction's + * memory context. This save the number of tuple memory contexts + * to create. + * + * XXX: this assumes that sub-transactions are freed (when being + * serialized or being streamed) together with its top-level + * transaction. + */ + txn = rbtxn_get_toptxn(txn); + + if (txn->concurrent_abort) + return NULL; + + return ReorderBufferGetTupleBufForTXN(rb, txn, tuple_len); +} + /* * Free a HeapTuple returned by ReorderBufferGetTupleBuf(). */ @@ -678,6 +706,7 @@ ReorderBufferTXNByXid(ReorderBuffer *rb, TransactionId xid, bool create, txn = ent->txn; txn->first_lsn = lsn; txn->restart_decoding_lsn = rb->current_restart_decoding_lsn; + txn->tup_context = NULL; if (create_as_top) { @@ -3765,6 +3794,10 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) txn->nentries_mem = 0; txn->txn_flags |= RBTXN_IS_SERIALIZED; + /* Remove the memory context for storing decoded tuples */ + if (txn->tup_context != NULL) + MemoryContextReset(txn->tup_context); + if (fd != -1) CloseTransientFile(fd); } @@ -4396,7 +4429,8 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn, uint32 tuplelen = ((HeapTuple) data)->t_len; change->data.tp.oldtuple = - ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader); + ReorderBufferGetTupleBufForTXN(rb, txn, + tuplelen - SizeofHeapTupleHeader); /* restore ->tuple */ memcpy(change->data.tp.oldtuple, data, @@ -4421,7 +4455,8 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn, sizeof(uint32)); change->data.tp.newtuple = - ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader); + ReorderBufferGetTupleBufForTXN(rb, txn, + tuplelen - SizeofHeapTupleHeader); /* restore ->tuple */ memcpy(change->data.tp.newtuple, data, diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 851a001c8b..ab3350ca51 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -407,6 +407,9 @@ typedef struct ReorderBufferTXN */ pairingheap_node txn_node; + /* Memory context for storing decoded tuples */ + MemoryContext tup_context; + /* * Size of this transaction (changes currently in memory, in bytes). */ @@ -626,7 +629,6 @@ struct ReorderBuffer */ MemoryContext change_context; MemoryContext txn_context; - MemoryContext tup_context; XLogRecPtr current_restart_decoding_lsn; @@ -668,9 +670,14 @@ struct ReorderBuffer extern ReorderBuffer *ReorderBufferAllocate(void); extern void ReorderBufferFree(ReorderBuffer *rb); -extern HeapTuple ReorderBufferGetTupleBuf(ReorderBuffer *rb, - Size tuple_len); extern void ReorderBufferReturnTupleBuf(HeapTuple tuple); +extern HeapTuple ReorderBufferGetTupleBufForTXN(ReorderBuffer *rb, + ReorderBufferTXN *txn, + Size tuple_len); +extern HeapTuple ReorderBufferGetTupleBufForXid(ReorderBuffer *rb, + TransactionId xid, + XLogRecPtr lsn, + Size tuple_len); extern ReorderBufferChange *ReorderBufferGetChange(ReorderBuffer *rb); extern void ReorderBufferReturnChange(ReorderBuffer *rb,