open-dsi
/
postgres
mirror of https://github.com/postgres/postgres
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove ReorderBufferTupleBuf structure.

Browse Source 
Since commit a4ccc1cef, the 'node' and 'alloc_tuple_size' fields of
the ReorderBufferTupleBuf structure are no longer used. This leaves
only the 'tuple' field in the structure. Since keeping a single-field
structure makes little sense, the ReorderBufferTupleBuf is removed
entirely. The code is refactored accordingly.

No back-patching since these are ABI changes in an exposed structure
and functions, and there would be some risk of breaking extensions.

Author: Aleksander Alekseev
Reviewed-by: Amit Kapila, Masahiko Sawada, Reid Thompson
Discussion: https://postgr.es/m/CAD21AoCvnuxiXXfRecp7g9+CeC35POQfhuQeJFr7_9u_Q5jc_Q@mail.gmail.com
pull/154/head
Masahiko Sawada 1 year ago
parent 50b797dc99
commit 08e6344fd6
6 changed files with 68 additions and 91 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 8
      contrib/test_decoding/test_decoding.c
  2. 26
      src/backend/replication/logical/decode.c
  3. 89
      src/backend/replication/logical/reorderbuffer.c
  4. 4
      src/backend/replication/pgoutput/pgoutput.c
  5. 31
      src/include/replication/reorderbuffer.h
  6. 1
      src/tools/pgindent/typedefs.list

8
contrib/test_decoding/test_decoding.c
Unescape View File

@ -640,7 +640,7 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
appendStringInfoString(ctx->out, " (no-tuple-data)");
else
tuple_to_stringinfo(ctx->out, tupdesc,
&change->data.tp.newtuple->tuple,
change->data.tp.newtuple,
false);
break;
case REORDER_BUFFER_CHANGE_UPDATE:
@ -649,7 +649,7 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
{
appendStringInfoString(ctx->out, " old-key:");
tuple_to_stringinfo(ctx->out, tupdesc,
&change->data.tp.oldtuple->tuple,
change->data.tp.oldtuple,
true);
appendStringInfoString(ctx->out, " new-tuple:");
}
@ -658,7 +658,7 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
appendStringInfoString(ctx->out, " (no-tuple-data)");
else
tuple_to_stringinfo(ctx->out, tupdesc,
&change->data.tp.newtuple->tuple,
change->data.tp.newtuple,
false);
break;
case REORDER_BUFFER_CHANGE_DELETE:
@ -670,7 +670,7 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* In DELETE, only the replica identity is present; display that */
else
tuple_to_stringinfo(ctx->out, tupdesc,
&change->data.tp.oldtuple->tuple,
change->data.tp.oldtuple,
true);
break;
default:

26
src/backend/replication/logical/decode.c
Unescape View File

@ -62,7 +62,7 @@ static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
/* common function to decode tuples */
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tuple);
static void DecodeXLogTuple(char *data, Size len, HeapTuple tuple);
/* helper functions for decoding transactions */
static inline bool FilterPrepare(LogicalDecodingContext *ctx,
@ -1152,7 +1152,7 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
ReorderBufferChange *change;
xl_multi_insert_tuple *xlhdr;
int datalen;
ReorderBufferTupleBuf *tuple;
HeapTuple tuple;
HeapTupleHeader header;
change = ReorderBufferGetChange(ctx->reorder);
@ -1169,21 +1169,21 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
ReorderBufferGetTupleBuf(ctx->reorder, datalen);
tuple = change->data.tp.newtuple;
header = tuple->tuple.t_data;
header = tuple->t_data;
/* not a disk based tuple */
ItemPointerSetInvalid(&tuple->tuple.t_self);
ItemPointerSetInvalid(&tuple->t_self);
/*
* We can only figure this out after reassembling the transactions.
*/
tuple->tuple.t_tableOid = InvalidOid;
tuple->t_tableOid = InvalidOid;
tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
tuple->t_len = datalen + SizeofHeapTupleHeader;
memset(header, 0, SizeofHeapTupleHeader);
memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
memcpy((char *) tuple->t_data + SizeofHeapTupleHeader,
(char *) data,
datalen);
header->t_infomask = xlhdr->t_infomask;
@ -1253,7 +1253,7 @@ DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
* computed outside as they are record specific.
*/
static void
DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tuple)
DecodeXLogTuple(char *data, Size len, HeapTuple tuple)
{
xl_heap_header xlhdr;
int datalen = len - SizeOfHeapHeader;
@ -1261,14 +1261,14 @@ DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tuple)
Assert(datalen >= 0);
tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
header = tuple->tuple.t_data;
tuple->t_len = datalen + SizeofHeapTupleHeader;
header = tuple->t_data;
/* not a disk based tuple */
ItemPointerSetInvalid(&tuple->tuple.t_self);
ItemPointerSetInvalid(&tuple->t_self);
/* we can only figure this out after reassembling the transactions */
tuple->tuple.t_tableOid = InvalidOid;
tuple->t_tableOid = InvalidOid;
/* data is not stored aligned, copy to aligned storage */
memcpy((char *) &xlhdr,
@ -1277,7 +1277,7 @@ DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tuple)
memset(header, 0, SizeofHeapTupleHeader);
memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
memcpy(((char *) tuple->t_data) + SizeofHeapTupleHeader,
data + SizeOfHeapHeader,
datalen);

89
src/backend/replication/logical/reorderbuffer.c
Unescape View File

@ -498,13 +498,13 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change,
case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
if (change->data.tp.newtuple)
{
ReorderBufferReturnTupleBuf(rb, change->data.tp.newtuple);
ReorderBufferReturnTupleBuf(change->data.tp.newtuple);
change->data.tp.newtuple = NULL;
}
if (change->data.tp.oldtuple)
{
ReorderBufferReturnTupleBuf(rb, change->data.tp.oldtuple);
ReorderBufferReturnTupleBuf(change->data.tp.oldtuple);
change->data.tp.oldtuple = NULL;
}
break;
@ -547,32 +547,29 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change,
}
/*
* Get a fresh ReorderBufferTupleBuf fitting at least a tuple of size
* tuple_len (excluding header overhead).
* Get a fresh HeapTuple fitting a tuple of size tuple_len (excluding header
* overhead).
*/
ReorderBufferTupleBuf *
HeapTuple
ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)
{
ReorderBufferTupleBuf *tuple;
HeapTuple tuple;
Size alloc_len;
alloc_len = tuple_len + SizeofHeapTupleHeader;
tuple = (ReorderBufferTupleBuf *)
MemoryContextAlloc(rb->tup_context,
sizeof(ReorderBufferTupleBuf) +
MAXIMUM_ALIGNOF + alloc_len);
tuple->alloc_tuple_size = alloc_len;
tuple->tuple.t_data = ReorderBufferTupleBufData(tuple);
tuple = (HeapTuple) MemoryContextAlloc(rb->tup_context,
HEAPTUPLESIZE + alloc_len);
tuple->t_data = (HeapTupleHeader) ((char *) tuple + HEAPTUPLESIZE);
return tuple;
}
/*
* Free a ReorderBufferTupleBuf.
* Free a HeapTuple returned by ReorderBufferGetTupleBuf().
*/
void
ReorderBufferReturnTupleBuf(ReorderBuffer *rb, ReorderBufferTupleBuf *tuple)
ReorderBufferReturnTupleBuf(HeapTuple tuple)
{
pfree(tuple);
}
@ -3759,8 +3756,8 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
{
char *data;
ReorderBufferTupleBuf *oldtup,
*newtup;
HeapTuple oldtup,
newtup;
Size oldlen = 0;
Size newlen = 0;
@ -3770,14 +3767,14 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
if (oldtup)
{
sz += sizeof(HeapTupleData);
oldlen = oldtup->tuple.t_len;
oldlen = oldtup->t_len;
sz += oldlen;
}
if (newtup)
{
sz += sizeof(HeapTupleData);
newlen = newtup->tuple.t_len;
newlen = newtup->t_len;
sz += newlen;
}
@ -3790,19 +3787,19 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
if (oldlen)
{
memcpy(data, &oldtup->tuple, sizeof(HeapTupleData));
memcpy(data, oldtup, sizeof(HeapTupleData));
data += sizeof(HeapTupleData);
memcpy(data, oldtup->tuple.t_data, oldlen);
memcpy(data, oldtup->t_data, oldlen);
data += oldlen;
}
if (newlen)
{
memcpy(data, &newtup->tuple, sizeof(HeapTupleData));
memcpy(data, newtup, sizeof(HeapTupleData));
data += sizeof(HeapTupleData);
memcpy(data, newtup->tuple.t_data, newlen);
memcpy(data, newtup->t_data, newlen);
data += newlen;
}
break;
@ -4118,8 +4115,8 @@ ReorderBufferChangeSize(ReorderBufferChange *change)
case REORDER_BUFFER_CHANGE_DELETE:
case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
{
ReorderBufferTupleBuf *oldtup,
*newtup;
HeapTuple oldtup,
newtup;
Size oldlen = 0;
Size newlen = 0;
@ -4129,14 +4126,14 @@ ReorderBufferChangeSize(ReorderBufferChange *change)
if (oldtup)
{
sz += sizeof(HeapTupleData);
oldlen = oldtup->tuple.t_len;
oldlen = oldtup->t_len;
sz += oldlen;
}
if (newtup)
{
sz += sizeof(HeapTupleData);
newlen = newtup->tuple.t_len;
newlen = newtup->t_len;
sz += newlen;
}
@ -4365,16 +4362,16 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader);
/* restore ->tuple */
memcpy(&change->data.tp.oldtuple->tuple, data,
memcpy(change->data.tp.oldtuple, data,
sizeof(HeapTupleData));
data += sizeof(HeapTupleData);
/* reset t_data pointer into the new tuplebuf */
change->data.tp.oldtuple->tuple.t_data =
ReorderBufferTupleBufData(change->data.tp.oldtuple);
change->data.tp.oldtuple->t_data =
(HeapTupleHeader) ((char *) change->data.tp.oldtuple + HEAPTUPLESIZE);
/* restore tuple data itself */
memcpy(change->data.tp.oldtuple->tuple.t_data, data, tuplelen);
memcpy(change->data.tp.oldtuple->t_data, data, tuplelen);
data += tuplelen;
}
@ -4390,16 +4387,16 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader);
/* restore ->tuple */
memcpy(&change->data.tp.newtuple->tuple, data,
memcpy(change->data.tp.newtuple, data,
sizeof(HeapTupleData));
data += sizeof(HeapTupleData);
/* reset t_data pointer into the new tuplebuf */
change->data.tp.newtuple->tuple.t_data =
ReorderBufferTupleBufData(change->data.tp.newtuple);
change->data.tp.newtuple->t_data =
(HeapTupleHeader) ((char *) change->data.tp.newtuple + HEAPTUPLESIZE);
/* restore tuple data itself */
memcpy(change->data.tp.newtuple->tuple.t_data, data, tuplelen);
memcpy(change->data.tp.newtuple->t_data, data, tuplelen);
data += tuplelen;
}
@ -4646,7 +4643,7 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
Relation relation, ReorderBufferChange *change)
{
ReorderBufferToastEnt *ent;
ReorderBufferTupleBuf *newtup;
HeapTuple newtup;
bool found;
int32 chunksize;
bool isnull;
@ -4661,9 +4658,9 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
Assert(IsToastRelation(relation));
newtup = change->data.tp.newtuple;
chunk_id = DatumGetObjectId(fastgetattr(&newtup->tuple, 1, desc, &isnull));
chunk_id = DatumGetObjectId(fastgetattr(newtup, 1, desc, &isnull));
Assert(!isnull);
chunk_seq = DatumGetInt32(fastgetattr(&newtup->tuple, 2, desc, &isnull));
chunk_seq = DatumGetInt32(fastgetattr(newtup, 2, desc, &isnull));
Assert(!isnull);
ent = (ReorderBufferToastEnt *)
@ -4686,7 +4683,7 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq %d",
chunk_seq, chunk_id, ent->last_chunk_seq + 1);
chunk = DatumGetPointer(fastgetattr(&newtup->tuple, 3, desc, &isnull));
chunk = DatumGetPointer(fastgetattr(newtup, 3, desc, &isnull));
Assert(!isnull);
/* calculate size so we can allocate the right size at once later */
@ -4737,7 +4734,7 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
Relation toast_rel;
TupleDesc toast_desc;
MemoryContext oldcontext;
ReorderBufferTupleBuf *newtup;
HeapTuple newtup;
Size old_size;
/* no toast tuples changed */
@ -4777,7 +4774,7 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
newtup = change->data.tp.newtuple;
heap_deform_tuple(&newtup->tuple, desc, attrs, isnull);
heap_deform_tuple(newtup, desc, attrs, isnull);
for (natt = 0; natt < desc->natts; natt++)
{
@ -4842,12 +4839,12 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
{
bool cisnull;
ReorderBufferChange *cchange;
ReorderBufferTupleBuf *ctup;
HeapTuple ctup;
Pointer chunk;
cchange = dlist_container(ReorderBufferChange, node, it.cur);
ctup = cchange->data.tp.newtuple;
chunk = DatumGetPointer(fastgetattr(&ctup->tuple, 3, toast_desc, &cisnull));
chunk = DatumGetPointer(fastgetattr(ctup, 3, toast_desc, &cisnull));
Assert(!cisnull);
Assert(!VARATT_IS_EXTERNAL(chunk));
@ -4882,11 +4879,11 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
* the tuplebuf because attrs[] will point back into the current content.
*/
tmphtup = heap_form_tuple(desc, attrs, isnull);
Assert(newtup->tuple.t_len <= MaxHeapTupleSize);
Assert(ReorderBufferTupleBufData(newtup) == newtup->tuple.t_data);
Assert(newtup->t_len <= MaxHeapTupleSize);
Assert(newtup->t_data == (HeapTupleHeader) ((char *) newtup + HEAPTUPLESIZE));
memcpy(newtup->tuple.t_data, tmphtup->t_data, tmphtup->t_len);
newtup->tuple.t_len = tmphtup->t_len;
memcpy(newtup->t_data, tmphtup->t_data, tmphtup->t_len);
newtup->t_len = tmphtup->t_len;
/*
* free resources we won't further need, more persistent stuff will be

4
src/backend/replication/pgoutput/pgoutput.c
Unescape View File

@ -1473,7 +1473,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
if (change->data.tp.oldtuple)
{
old_slot = relentry->old_slot;
ExecStoreHeapTuple(&change->data.tp.oldtuple->tuple, old_slot, false);
ExecStoreHeapTuple(change->data.tp.oldtuple, old_slot, false);
/* Convert tuple if needed. */
if (relentry->attrmap)
@ -1488,7 +1488,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
if (change->data.tp.newtuple)
{
new_slot = relentry->new_slot;
ExecStoreHeapTuple(&change->data.tp.newtuple->tuple, new_slot, false);
ExecStoreHeapTuple(change->data.tp.newtuple, new_slot, false);
/* Convert tuple if needed. */
if (relentry->attrmap)

31
src/include/replication/reorderbuffer.h
Unescape View File

@ -28,25 +28,6 @@ typedef enum
DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE,
} DebugLogicalRepStreamingMode;
/* an individual tuple, stored in one chunk of memory */
typedef struct ReorderBufferTupleBuf
{
/* position in preallocated list */
slist_node node;
/* tuple header, the interesting bit for users of logical decoding */
HeapTupleData tuple;
/* pre-allocated size of tuple buffer, different from tuple size */
Size alloc_tuple_size;
/* actual tuple data follows */
} ReorderBufferTupleBuf;
/* pointer to the data stored in a TupleBuf */
#define ReorderBufferTupleBufData(p) \
((HeapTupleHeader) MAXALIGN(((char *) p) + sizeof(ReorderBufferTupleBuf)))
/*
* Types of the change passed to a 'change' callback.
*
@ -114,9 +95,9 @@ typedef struct ReorderBufferChange
bool clear_toast_afterwards;
/* valid for DELETE || UPDATE */
ReorderBufferTupleBuf *oldtuple;
HeapTuple oldtuple;
/* valid for INSERT || UPDATE */
ReorderBufferTupleBuf *newtuple;
HeapTuple newtuple;
} tp;
/*
@ -678,10 +659,10 @@ struct ReorderBuffer
extern ReorderBuffer *ReorderBufferAllocate(void);
extern void ReorderBufferFree(ReorderBuffer *rb);
extern ReorderBufferTupleBuf *ReorderBufferGetTupleBuf(ReorderBuffer *rb,
Size tuple_len);
extern void ReorderBufferReturnTupleBuf(ReorderBuffer *rb,
ReorderBufferTupleBuf *tuple);
extern HeapTuple ReorderBufferGetTupleBuf(ReorderBuffer *rb,
Size tuple_len);
extern void ReorderBufferReturnTupleBuf(HeapTuple tuple);
extern ReorderBufferChange *ReorderBufferGetChange(ReorderBuffer *rb);
extern void ReorderBufferReturnChange(ReorderBuffer *rb,
ReorderBufferChange *change, bool upd_mem);

1
src/tools/pgindent/typedefs.list
Unescape View File

@ -2351,7 +2351,6 @@ ReorderBufferStreamTruncateCB
ReorderBufferTXN
ReorderBufferTXNByIdEnt
ReorderBufferToastEnt
ReorderBufferTupleBuf
ReorderBufferTupleCidEnt
ReorderBufferTupleCidKey
ReorderBufferUpdateProgressTxnCB

Write Preview
Loading…
Cancel
Save