database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-07-31 22:04:40 +03:00
Code Activity

Remove unportable use of anonymous unions from reorderbuffer.h.

Browse Source 
In b89e151054 I had assumed it was ok to use anonymous unions as
struct members, but while a longstanding extension in many compilers,
it's only been standardized in C11.

To fix, remove one of the anonymous unions which tried to hide some
implementation specific enum values and give the other a name. The
latter unfortunately requires changes in output plugins, but since the
feature has only been added a few days ago...

Andres Freund
This commit is contained in:
Tom Lane
2014-03-07 17:02:48 -05:00
parent 83204e100c
commit ea177a3ba7
4 changed files with 183 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
contrib/test_decoding/test_decoding.c
View File

@ -358,43 +358,45 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
{ {
case REORDER_BUFFER_CHANGE_INSERT: case REORDER_BUFFER_CHANGE_INSERT:
appendStringInfoString(ctx->out, " INSERT:"); appendStringInfoString(ctx->out, " INSERT:");
if (change->tp.newtuple == NULL) if (change->data.tp.newtuple == NULL)
appendStringInfoString(ctx->out, " (no-tuple-data)"); appendStringInfoString(ctx->out, " (no-tuple-data)");
else else
tuple_to_stringinfo(ctx->out, tupdesc, tuple_to_stringinfo(ctx->out, tupdesc,
&change->tp.newtuple->tuple, &change->data.tp.newtuple->tuple,
false); false);
break; break;
case REORDER_BUFFER_CHANGE_UPDATE: case REORDER_BUFFER_CHANGE_UPDATE:
appendStringInfoString(ctx->out, " UPDATE:"); appendStringInfoString(ctx->out, " UPDATE:");
if (change->tp.oldtuple != NULL) if (change->data.tp.oldtuple != NULL)
{ {
appendStringInfoString(ctx->out, " old-key:"); appendStringInfoString(ctx->out, " old-key:");
tuple_to_stringinfo(ctx->out, tupdesc, tuple_to_stringinfo(ctx->out, tupdesc,
&change->tp.oldtuple->tuple, &change->data.tp.oldtuple->tuple,
true); true);
appendStringInfoString(ctx->out, " new-tuple:"); appendStringInfoString(ctx->out, " new-tuple:");
} }
if (change->tp.newtuple == NULL) if (change->data.tp.newtuple == NULL)
appendStringInfoString(ctx->out, " (no-tuple-data)"); appendStringInfoString(ctx->out, " (no-tuple-data)");
else else
tuple_to_stringinfo(ctx->out, tupdesc, tuple_to_stringinfo(ctx->out, tupdesc,
&change->tp.newtuple->tuple, &change->data.tp.newtuple->tuple,
false); false);
break; break;
case REORDER_BUFFER_CHANGE_DELETE: case REORDER_BUFFER_CHANGE_DELETE:
appendStringInfoString(ctx->out, " DELETE:"); appendStringInfoString(ctx->out, " DELETE:");
/* if there was no PK, we only know that a delete happened */ /* if there was no PK, we only know that a delete happened */
if (change->tp.oldtuple == NULL) if (change->data.tp.oldtuple == NULL)
appendStringInfoString(ctx->out, " (no-tuple-data)"); appendStringInfoString(ctx->out, " (no-tuple-data)");
/* In DELETE, only the replica identity is present; display that */ /* In DELETE, only the replica identity is present; display that */
else else
tuple_to_stringinfo(ctx->out, tupdesc, tuple_to_stringinfo(ctx->out, tupdesc,
&change->tp.oldtuple->tuple, &change->data.tp.oldtuple->tuple,
true); true);
break; break;
default:
Assert(false);
} }
MemoryContextSwitchTo(old); MemoryContextSwitchTo(old);

28
src/backend/replication/logical/decode.c
View File

@ -586,17 +586,17 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
change = ReorderBufferGetChange(ctx->reorder); change = ReorderBufferGetChange(ctx->reorder);
change->action = REORDER_BUFFER_CHANGE_INSERT; change->action = REORDER_BUFFER_CHANGE_INSERT;
memcpy(&change->tp.relnode, &xlrec->target.node, sizeof(RelFileNode)); memcpy(&change->data.tp.relnode, &xlrec->target.node, sizeof(RelFileNode));
if (xlrec->flags & XLOG_HEAP_CONTAINS_NEW_TUPLE) if (xlrec->flags & XLOG_HEAP_CONTAINS_NEW_TUPLE)
{ {
Assert(r->xl_len > (SizeOfHeapInsert + SizeOfHeapHeader)); Assert(r->xl_len > (SizeOfHeapInsert + SizeOfHeapHeader));
change->tp.newtuple = ReorderBufferGetTupleBuf(ctx->reorder); change->data.tp.newtuple = ReorderBufferGetTupleBuf(ctx->reorder);
DecodeXLogTuple((char *) xlrec + SizeOfHeapInsert, DecodeXLogTuple((char *) xlrec + SizeOfHeapInsert,
r->xl_len - SizeOfHeapInsert, r->xl_len - SizeOfHeapInsert,
change->tp.newtuple); change->data.tp.newtuple);
} }
ReorderBufferQueueChange(ctx->reorder, r->xl_xid, buf->origptr, change); ReorderBufferQueueChange(ctx->reorder, r->xl_xid, buf->origptr, change);
@ -626,7 +626,7 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
change = ReorderBufferGetChange(ctx->reorder); change = ReorderBufferGetChange(ctx->reorder);
change->action = REORDER_BUFFER_CHANGE_UPDATE; change->action = REORDER_BUFFER_CHANGE_UPDATE;
memcpy(&change->tp.relnode, &xlrec->target.node, sizeof(RelFileNode)); memcpy(&change->data.tp.relnode, &xlrec->target.node, sizeof(RelFileNode));
data = (char *) &xlhdr->header; data = (char *) &xlhdr->header;
@ -634,11 +634,11 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
{ {
Assert(r->xl_len > (SizeOfHeapUpdate + SizeOfHeapHeaderLen)); Assert(r->xl_len > (SizeOfHeapUpdate + SizeOfHeapHeaderLen));
change->tp.newtuple = ReorderBufferGetTupleBuf(ctx->reorder); change->data.tp.newtuple = ReorderBufferGetTupleBuf(ctx->reorder);
DecodeXLogTuple(data, DecodeXLogTuple(data,
xlhdr->t_len + SizeOfHeapHeader, xlhdr->t_len + SizeOfHeapHeader,
change->tp.newtuple); change->data.tp.newtuple);
/* skip over the rest of the tuple header */ /* skip over the rest of the tuple header */
data += SizeOfHeapHeader; data += SizeOfHeapHeader;
/* skip over the tuple data */ /* skip over the tuple data */
@ -648,10 +648,10 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
if (xlrec->flags & XLOG_HEAP_CONTAINS_OLD) if (xlrec->flags & XLOG_HEAP_CONTAINS_OLD)
{ {
xlhdr = (xl_heap_header_len *) data; xlhdr = (xl_heap_header_len *) data;
change->tp.oldtuple = ReorderBufferGetTupleBuf(ctx->reorder); change->data.tp.oldtuple = ReorderBufferGetTupleBuf(ctx->reorder);
DecodeXLogTuple((char *) &xlhdr->header, DecodeXLogTuple((char *) &xlhdr->header,
xlhdr->t_len + SizeOfHeapHeader, xlhdr->t_len + SizeOfHeapHeader,
change->tp.oldtuple); change->data.tp.oldtuple);
data = (char *) &xlhdr->header; data = (char *) &xlhdr->header;
data += SizeOfHeapHeader; data += SizeOfHeapHeader;
data += xlhdr->t_len; data += xlhdr->t_len;
@ -681,18 +681,18 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
change = ReorderBufferGetChange(ctx->reorder); change = ReorderBufferGetChange(ctx->reorder);
change->action = REORDER_BUFFER_CHANGE_DELETE; change->action = REORDER_BUFFER_CHANGE_DELETE;
memcpy(&change->tp.relnode, &xlrec->target.node, sizeof(RelFileNode)); memcpy(&change->data.tp.relnode, &xlrec->target.node, sizeof(RelFileNode));
/* old primary key stored */ /* old primary key stored */
if (xlrec->flags & XLOG_HEAP_CONTAINS_OLD) if (xlrec->flags & XLOG_HEAP_CONTAINS_OLD)
{ {
Assert(r->xl_len > (SizeOfHeapDelete + SizeOfHeapHeader)); Assert(r->xl_len > (SizeOfHeapDelete + SizeOfHeapHeader));
change->tp.oldtuple = ReorderBufferGetTupleBuf(ctx->reorder); change->data.tp.oldtuple = ReorderBufferGetTupleBuf(ctx->reorder);
DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete, DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
r->xl_len - SizeOfHeapDelete, r->xl_len - SizeOfHeapDelete,
change->tp.oldtuple); change->data.tp.oldtuple);
} }
ReorderBufferQueueChange(ctx->reorder, r->xl_xid, buf->origptr, change); ReorderBufferQueueChange(ctx->reorder, r->xl_xid, buf->origptr, change);
} }
@ -735,7 +735,7 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
change = ReorderBufferGetChange(ctx->reorder); change = ReorderBufferGetChange(ctx->reorder);
change->action = REORDER_BUFFER_CHANGE_INSERT; change->action = REORDER_BUFFER_CHANGE_INSERT;
memcpy(&change->tp.relnode, &xlrec->node, sizeof(RelFileNode)); memcpy(&change->data.tp.relnode, &xlrec->node, sizeof(RelFileNode));
/* /*
* CONTAINS_NEW_TUPLE will always be set currently as multi_insert * CONTAINS_NEW_TUPLE will always be set currently as multi_insert
@ -746,9 +746,9 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
*/ */
if (xlrec->flags & XLOG_HEAP_CONTAINS_NEW_TUPLE) if (xlrec->flags & XLOG_HEAP_CONTAINS_NEW_TUPLE)
{ {
change->tp.newtuple = ReorderBufferGetTupleBuf(ctx->reorder); change->data.tp.newtuple = ReorderBufferGetTupleBuf(ctx->reorder);
tuple = change->tp.newtuple; tuple = change->data.tp.newtuple;
/* not a disk based tuple */ /* not a disk based tuple */
ItemPointerSetInvalid(&tuple->tuple.t_self); ItemPointerSetInvalid(&tuple->tuple.t_self);

277
src/backend/replication/logical/reorderbuffer.c
View File

@ -80,26 +80,6 @@
#include "utils/relfilenodemap.h" #include "utils/relfilenodemap.h"
#include "utils/tqual.h" #include "utils/tqual.h"
/*
* For efficiency and simplicity reasons we want to keep Snapshots, CommandIds
* and ComboCids in the same list with the user visible INSERT/UPDATE/DELETE
* changes. We don't want to leak those internal values to external users
* though (they would just use switch()...default:) because that would make it
* harder to add to new user visible values.
*
* This needs to be synchronized with ReorderBufferChangeType! Adjust the
* StaticAssertExpr's in ReorderBufferAllocate if you add anything!
*/
typedef enum
{
REORDER_BUFFER_CHANGE_INTERNAL_INSERT,
REORDER_BUFFER_CHANGE_INTERNAL_UPDATE,
REORDER_BUFFER_CHANGE_INTERNAL_DELETE,
REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT,
REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID,
REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID
} ReorderBufferChangeTypeInternal;
/* entry for a hash table we use to map from xid to our transaction state */ /* entry for a hash table we use to map from xid to our transaction state */
typedef struct ReorderBufferTXNByIdEnt typedef struct ReorderBufferTXNByIdEnt
{ {
@ -255,10 +235,6 @@ ReorderBufferAllocate(void)
HASHCTL hash_ctl; HASHCTL hash_ctl;
MemoryContext new_ctx; MemoryContext new_ctx;
StaticAssertExpr((int) REORDER_BUFFER_CHANGE_INTERNAL_INSERT == (int) REORDER_BUFFER_CHANGE_INSERT, "out of sync enums");
StaticAssertExpr((int) REORDER_BUFFER_CHANGE_INTERNAL_UPDATE == (int) REORDER_BUFFER_CHANGE_UPDATE, "out of sync enums");
StaticAssertExpr((int) REORDER_BUFFER_CHANGE_INTERNAL_DELETE == (int) REORDER_BUFFER_CHANGE_DELETE, "out of sync enums");
/* allocate memory in own context, to have better accountability */ /* allocate memory in own context, to have better accountability */
new_ctx = AllocSetContextCreate(CurrentMemoryContext, new_ctx = AllocSetContextCreate(CurrentMemoryContext,
"ReorderBuffer", "ReorderBuffer",
@ -427,28 +403,28 @@ void
ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change) ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change)
{ {
/* free contained data */ /* free contained data */
switch ((ReorderBufferChangeTypeInternal) change->action_internal) switch (change->action)
{ {
case REORDER_BUFFER_CHANGE_INTERNAL_INSERT: case REORDER_BUFFER_CHANGE_INSERT:
case REORDER_BUFFER_CHANGE_INTERNAL_UPDATE: case REORDER_BUFFER_CHANGE_UPDATE:
case REORDER_BUFFER_CHANGE_INTERNAL_DELETE: case REORDER_BUFFER_CHANGE_DELETE:
if (change->tp.newtuple) if (change->data.tp.newtuple)
{ {
ReorderBufferReturnTupleBuf(rb, change->tp.newtuple); ReorderBufferReturnTupleBuf(rb, change->data.tp.newtuple);
change->tp.newtuple = NULL; change->data.tp.newtuple = NULL;
} }
if (change->tp.oldtuple) if (change->data.tp.oldtuple)
{ {
ReorderBufferReturnTupleBuf(rb, change->tp.oldtuple); ReorderBufferReturnTupleBuf(rb, change->data.tp.oldtuple);
change->tp.oldtuple = NULL; change->data.tp.oldtuple = NULL;
} }
break; break;
case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT: case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
if (change->snapshot) if (change->data.snapshot)
{ {
ReorderBufferFreeSnap(rb, change->snapshot); ReorderBufferFreeSnap(rb, change->data.snapshot);
change->snapshot = NULL; change->data.snapshot = NULL;
} }
break; break;
case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID: case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
@ -1086,7 +1062,7 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
ReorderBufferChange *change; ReorderBufferChange *change;
change = dlist_container(ReorderBufferChange, node, iter.cur); change = dlist_container(ReorderBufferChange, node, iter.cur);
Assert(change->action_internal == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID); Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
ReorderBufferReturnChange(rb, change); ReorderBufferReturnChange(rb, change);
} }
@ -1161,14 +1137,14 @@ ReorderBufferBuildTupleCidHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
change = dlist_container(ReorderBufferChange, node, iter.cur); change = dlist_container(ReorderBufferChange, node, iter.cur);
Assert(change->action_internal == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID); Assert(change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
/* be careful about padding */ /* be careful about padding */
memset(&key, 0, sizeof(ReorderBufferTupleCidKey)); memset(&key, 0, sizeof(ReorderBufferTupleCidKey));
key.relnode = change->tuplecid.node; key.relnode = change->data.tuplecid.node;
ItemPointerCopy(&change->tuplecid.tid, ItemPointerCopy(&change->data.tuplecid.tid,
&key.tid); &key.tid);
ent = (ReorderBufferTupleCidEnt *) ent = (ReorderBufferTupleCidEnt *)
@ -1178,22 +1154,22 @@ ReorderBufferBuildTupleCidHash(ReorderBuffer *rb, ReorderBufferTXN *txn)
&found); &found);
if (!found) if (!found)
{ {
ent->cmin = change->tuplecid.cmin; ent->cmin = change->data.tuplecid.cmin;
ent->cmax = change->tuplecid.cmax; ent->cmax = change->data.tuplecid.cmax;
ent->combocid = change->tuplecid.combocid; ent->combocid = change->data.tuplecid.combocid;
} }
else else
{ {
Assert(ent->cmin == change->tuplecid.cmin); Assert(ent->cmin == change->data.tuplecid.cmin);
Assert(ent->cmax == InvalidCommandId || Assert(ent->cmax == InvalidCommandId ||
ent->cmax == change->tuplecid.cmax); ent->cmax == change->data.tuplecid.cmax);
/* /*
* if the tuple got valid in this transaction and now got deleted * if the tuple got valid in this transaction and now got deleted
* we already have a valid cmin stored. The cmax will be * we already have a valid cmin stored. The cmax will be
* InvalidCommandId though. * InvalidCommandId though.
*/ */
ent->cmax = change->tuplecid.cmax; ent->cmax = change->data.tuplecid.cmax;
} }
} }
} }
@ -1367,33 +1343,33 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
Relation relation = NULL; Relation relation = NULL;
Oid reloid; Oid reloid;
switch ((ReorderBufferChangeTypeInternal) change->action_internal) switch (change->action)
{ {
case REORDER_BUFFER_CHANGE_INTERNAL_INSERT: case REORDER_BUFFER_CHANGE_INSERT:
case REORDER_BUFFER_CHANGE_INTERNAL_UPDATE: case REORDER_BUFFER_CHANGE_UPDATE:
case REORDER_BUFFER_CHANGE_INTERNAL_DELETE: case REORDER_BUFFER_CHANGE_DELETE:
Assert(snapshot_now); Assert(snapshot_now);
reloid = RelidByRelfilenode(change->tp.relnode.spcNode, reloid = RelidByRelfilenode(change->data.tp.relnode.spcNode,
change->tp.relnode.relNode); change->data.tp.relnode.relNode);
/* /*
* Catalog tuple without data, emitted while catalog was * Catalog tuple without data, emitted while catalog was
* in the process of being rewritten. * in the process of being rewritten.
*/ */
if (reloid == InvalidOid && if (reloid == InvalidOid &&
change->tp.newtuple == NULL && change->data.tp.newtuple == NULL &&
change->tp.oldtuple == NULL) change->data.tp.oldtuple == NULL)
continue; continue;
else if (reloid == InvalidOid) else if (reloid == InvalidOid)
elog(ERROR, "could not lookup relation %s", elog(ERROR, "could not lookup relation %s",
relpathperm(change->tp.relnode, MAIN_FORKNUM)); relpathperm(change->data.tp.relnode, MAIN_FORKNUM));
relation = RelationIdGetRelation(reloid); relation = RelationIdGetRelation(reloid);
if (relation == NULL) if (relation == NULL)
elog(ERROR, "could open relation descriptor %s", elog(ERROR, "could open relation descriptor %s",
relpathperm(change->tp.relnode, MAIN_FORKNUM)); relpathperm(change->data.tp.relnode, MAIN_FORKNUM));
if (RelationIsLogicallyLogged(relation)) if (RelationIsLogicallyLogged(relation))
{ {
@ -1440,7 +1416,7 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
{ {
ReorderBufferFreeSnap(rb, snapshot_now); ReorderBufferFreeSnap(rb, snapshot_now);
snapshot_now = snapshot_now =
ReorderBufferCopySnap(rb, change->snapshot, ReorderBufferCopySnap(rb, change->data.snapshot,
txn, command_id); txn, command_id);
} }
/* /*
@ -1448,15 +1424,15 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
* free. We could introduce refcounting for that, but for * free. We could introduce refcounting for that, but for
* now this seems infrequent enough not to care. * now this seems infrequent enough not to care.
*/ */
else if (change->snapshot->copied) else if (change->data.snapshot->copied)
{ {
snapshot_now = snapshot_now =
ReorderBufferCopySnap(rb, change->snapshot, ReorderBufferCopySnap(rb, change->data.snapshot,
txn, command_id); txn, command_id);
} }
else else
{ {
snapshot_now = change->snapshot; snapshot_now = change->data.snapshot;
} }
@ -1465,11 +1441,11 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
break; break;
case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID: case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
Assert(change->command_id != InvalidCommandId); Assert(change->data.command_id != InvalidCommandId);
if (command_id < change->command_id) if (command_id < change->data.command_id)
{ {
command_id = change->command_id; command_id = change->data.command_id;
if (!snapshot_now->copied) if (!snapshot_now->copied)
{ {
@ -1712,8 +1688,8 @@ ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid,
{ {
ReorderBufferChange *change = ReorderBufferGetChange(rb); ReorderBufferChange *change = ReorderBufferGetChange(rb);
change->snapshot = snap; change->data.snapshot = snap;
change->action_internal = REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT; change->action = REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT;
ReorderBufferQueueChange(rb, xid, lsn, change); ReorderBufferQueueChange(rb, xid, lsn, change);
} }
@ -1752,8 +1728,8 @@ ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid,
{ {
ReorderBufferChange *change = ReorderBufferGetChange(rb); ReorderBufferChange *change = ReorderBufferGetChange(rb);
change->command_id = cid; change->data.command_id = cid;
change->action_internal = REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID; change->action = REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID;
ReorderBufferQueueChange(rb, xid, lsn, change); ReorderBufferQueueChange(rb, xid, lsn, change);
} }
@ -1773,13 +1749,13 @@ ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid,
txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true); txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
change->tuplecid.node = node; change->data.tuplecid.node = node;
change->tuplecid.tid = tid; change->data.tuplecid.tid = tid;
change->tuplecid.cmin = cmin; change->data.tuplecid.cmin = cmin;
change->tuplecid.cmax = cmax; change->data.tuplecid.cmax = cmax;
change->tuplecid.combocid = combocid; change->data.tuplecid.combocid = combocid;
change->lsn = lsn; change->lsn = lsn;
change->action_internal = REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID; change->action = REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID;
dlist_push_tail(&txn->tuplecids, &change->node); dlist_push_tail(&txn->tuplecids, &change->node);
txn->ntuplecids++; txn->ntuplecids++;
@ -2017,26 +1993,30 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
ondisk = (ReorderBufferDiskChange *) rb->outbuf; ondisk = (ReorderBufferDiskChange *) rb->outbuf;
memcpy(&ondisk->change, change, sizeof(ReorderBufferChange)); memcpy(&ondisk->change, change, sizeof(ReorderBufferChange));
switch ((ReorderBufferChangeTypeInternal) change->action_internal) switch (change->action)
{ {
case REORDER_BUFFER_CHANGE_INTERNAL_INSERT: case REORDER_BUFFER_CHANGE_INSERT:
/* fall through */ /* fall through */
case REORDER_BUFFER_CHANGE_INTERNAL_UPDATE: case REORDER_BUFFER_CHANGE_UPDATE:
/* fall through */ /* fall through */
case REORDER_BUFFER_CHANGE_INTERNAL_DELETE: case REORDER_BUFFER_CHANGE_DELETE:
{ {
char *data; char *data;
ReorderBufferTupleBuf *oldtup, *newtup;
Size oldlen = 0; Size oldlen = 0;
Size newlen = 0; Size newlen = 0;
if (change->tp.oldtuple) oldtup = change->data.tp.oldtuple;
newtup = change->data.tp.newtuple;
if (oldtup)
oldlen = offsetof(ReorderBufferTupleBuf, data) oldlen = offsetof(ReorderBufferTupleBuf, data)
+ change->tp.oldtuple->tuple.t_len + oldtup->tuple.t_len
- offsetof(HeapTupleHeaderData, t_bits); - offsetof(HeapTupleHeaderData, t_bits);
if (change->tp.newtuple) if (newtup)
newlen = offsetof(ReorderBufferTupleBuf, data) newlen = offsetof(ReorderBufferTupleBuf, data)
+ change->tp.newtuple->tuple.t_len + newtup->tuple.t_len
- offsetof(HeapTupleHeaderData, t_bits); - offsetof(HeapTupleHeaderData, t_bits);
sz += oldlen; sz += oldlen;
@ -2051,26 +2031,27 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
if (oldlen) if (oldlen)
{ {
memcpy(data, change->tp.oldtuple, oldlen); memcpy(data, oldtup, oldlen);
data += oldlen; data += oldlen;
Assert(&change->tp.oldtuple->header == change->tp.oldtuple->tuple.t_data);
} }
if (newlen) if (newlen)
{ {
memcpy(data, change->tp.newtuple, newlen); memcpy(data, newtup, newlen);
data += newlen; data += newlen;
Assert(&change->tp.newtuple->header == change->tp.newtuple->tuple.t_data);
} }
break; break;
} }
case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT: case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
{ {
Snapshot snap;
char *data; char *data;
snap = change->data.snapshot;
sz += sizeof(SnapshotData) + sz += sizeof(SnapshotData) +
sizeof(TransactionId) * change->snapshot->xcnt + sizeof(TransactionId) * snap->xcnt +
sizeof(TransactionId) * change->snapshot->subxcnt sizeof(TransactionId) * snap->subxcnt
; ;
/* make sure we have enough space */ /* make sure we have enough space */
@ -2079,21 +2060,21 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
/* might have been reallocated above */ /* might have been reallocated above */
ondisk = (ReorderBufferDiskChange *) rb->outbuf; ondisk = (ReorderBufferDiskChange *) rb->outbuf;
memcpy(data, change->snapshot, sizeof(SnapshotData)); memcpy(data, snap, sizeof(SnapshotData));
data += sizeof(SnapshotData); data += sizeof(SnapshotData);
if (change->snapshot->xcnt) if (snap->xcnt)
{ {
memcpy(data, change->snapshot->xip, memcpy(data, snap->xip,
sizeof(TransactionId) + change->snapshot->xcnt); sizeof(TransactionId) + snap->xcnt);
data += sizeof(TransactionId) + change->snapshot->xcnt; data += sizeof(TransactionId) + snap->xcnt;
} }
if (change->snapshot->subxcnt) if (snap->subxcnt)
{ {
memcpy(data, change->snapshot->subxip, memcpy(data, snap->subxip,
sizeof(TransactionId) + change->snapshot->subxcnt); sizeof(TransactionId) + snap->subxcnt);
data += sizeof(TransactionId) + change->snapshot->subxcnt; data += sizeof(TransactionId) + snap->subxcnt;
} }
break; break;
} }
@ -2116,7 +2097,7 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
txn->xid))); txn->xid)));
} }
Assert(ondisk->change.action_internal == change->action_internal); Assert(ondisk->change.action == change->action);
} }
/* /*
@ -2271,56 +2252,60 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
data += sizeof(ReorderBufferDiskChange); data += sizeof(ReorderBufferDiskChange);
/* restore individual stuff */ /* restore individual stuff */
switch ((ReorderBufferChangeTypeInternal) change->action_internal) switch (change->action)
{ {
case REORDER_BUFFER_CHANGE_INTERNAL_INSERT: case REORDER_BUFFER_CHANGE_INSERT:
/* fall through */ /* fall through */
case REORDER_BUFFER_CHANGE_INTERNAL_UPDATE: case REORDER_BUFFER_CHANGE_UPDATE:
/* fall through */ /* fall through */
case REORDER_BUFFER_CHANGE_INTERNAL_DELETE: case REORDER_BUFFER_CHANGE_DELETE:
if (change->tp.newtuple) if (change->data.tp.newtuple)
{ {
Size len = offsetof(ReorderBufferTupleBuf, data) Size len = offsetof(ReorderBufferTupleBuf, data)
+((ReorderBufferTupleBuf *) data)->tuple.t_len +((ReorderBufferTupleBuf *) data)->tuple.t_len
- offsetof(HeapTupleHeaderData, t_bits); - offsetof(HeapTupleHeaderData, t_bits);
change->tp.newtuple = ReorderBufferGetTupleBuf(rb); change->data.tp.newtuple = ReorderBufferGetTupleBuf(rb);
memcpy(change->tp.newtuple, data, len); memcpy(change->data.tp.newtuple, data, len);
change->tp.newtuple->tuple.t_data = &change->tp.newtuple->header; change->data.tp.newtuple->tuple.t_data =
&change->data.tp.newtuple->header;
data += len; data += len;
} }
if (change->tp.oldtuple) if (change->data.tp.oldtuple)
{ {
Size len = offsetof(ReorderBufferTupleBuf, data) Size len = offsetof(ReorderBufferTupleBuf, data)
+((ReorderBufferTupleBuf *) data)->tuple.t_len +((ReorderBufferTupleBuf *) data)->tuple.t_len
- offsetof(HeapTupleHeaderData, t_bits); - offsetof(HeapTupleHeaderData, t_bits);
change->tp.oldtuple = ReorderBufferGetTupleBuf(rb); change->data.tp.oldtuple = ReorderBufferGetTupleBuf(rb);
memcpy(change->tp.oldtuple, data, len); memcpy(change->data.tp.oldtuple, data, len);
change->tp.oldtuple->tuple.t_data = &change->tp.oldtuple->header; change->data.tp.oldtuple->tuple.t_data =
&change->data.tp.oldtuple->header;
data += len; data += len;
} }
break; break;
case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT: case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
{ {
Snapshot oldsnap = (Snapshot) data; Snapshot oldsnap;
Size size = sizeof(SnapshotData) + Snapshot newsnap;
sizeof(TransactionId) * oldsnap->xcnt + Size size;
sizeof(TransactionId) * (oldsnap->subxcnt + 0)
;
Assert(change->snapshot != NULL); oldsnap = (Snapshot) data;
change->snapshot = MemoryContextAllocZero(rb->context, size); size = sizeof(SnapshotData) +
sizeof(TransactionId) * oldsnap->xcnt +
sizeof(TransactionId) * (oldsnap->subxcnt + 0);
memcpy(change->snapshot, data, size); change->data.snapshot = MemoryContextAllocZero(rb->context, size);
change->snapshot->xip = (TransactionId *)
(((char *) change->snapshot) + sizeof(SnapshotData)); newsnap = change->data.snapshot;
change->snapshot->subxip =
change->snapshot->xip + change->snapshot->xcnt + 0; memcpy(newsnap, data, size);
change->snapshot->copied = true; newsnap->xip = (TransactionId *)
(((char *) newsnap) + sizeof(SnapshotData));
newsnap->subxip = newsnap->xip + newsnap->xcnt;
newsnap->copied = true;
break; break;
} }
/* the base struct contains all the data, easy peasy */ /* the base struct contains all the data, easy peasy */
@ -2464,6 +2449,7 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
Relation relation, ReorderBufferChange *change) Relation relation, ReorderBufferChange *change)
{ {
ReorderBufferToastEnt *ent; ReorderBufferToastEnt *ent;
ReorderBufferTupleBuf *newtup;
bool found; bool found;
int32 chunksize; int32 chunksize;
bool isnull; bool isnull;
@ -2477,9 +2463,10 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
Assert(IsToastRelation(relation)); Assert(IsToastRelation(relation));
chunk_id = DatumGetObjectId(fastgetattr(&change->tp.newtuple->tuple, 1, desc, &isnull)); newtup = change->data.tp.newtuple;
chunk_id = DatumGetObjectId(fastgetattr(&newtup->tuple, 1, desc, &isnull));
Assert(!isnull); Assert(!isnull);
chunk_seq = DatumGetInt32(fastgetattr(&change->tp.newtuple->tuple, 2, desc, &isnull)); chunk_seq = DatumGetInt32(fastgetattr(&newtup->tuple, 2, desc, &isnull));
Assert(!isnull); Assert(!isnull);
ent = (ReorderBufferToastEnt *) ent = (ReorderBufferToastEnt *)
@ -2505,7 +2492,7 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn,
elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq %d", elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq %d",
chunk_seq, chunk_id, ent->last_chunk_seq + 1); chunk_seq, chunk_id, ent->last_chunk_seq + 1);
chunk = DatumGetPointer(fastgetattr(&change->tp.newtuple->tuple, 3, desc, &isnull)); chunk = DatumGetPointer(fastgetattr(&newtup->tuple, 3, desc, &isnull));
Assert(!isnull); Assert(!isnull);
/* calculate size so we can allocate the right size at once later */ /* calculate size so we can allocate the right size at once later */
@ -2539,10 +2526,11 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
Datum *attrs; Datum *attrs;
bool *isnull; bool *isnull;
bool *free; bool *free;
HeapTuple newtup; HeapTuple tmphtup;
Relation toast_rel; Relation toast_rel;
TupleDesc toast_desc; TupleDesc toast_desc;
MemoryContext oldcontext; MemoryContext oldcontext;
ReorderBufferTupleBuf *newtup;
/* no toast tuples changed */ /* no toast tuples changed */
if (txn->toast_hash == NULL) if (txn->toast_hash == NULL)
@ -2551,7 +2539,7 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
oldcontext = MemoryContextSwitchTo(rb->context); oldcontext = MemoryContextSwitchTo(rb->context);
/* we should only have toast tuples in an INSERT or UPDATE */ /* we should only have toast tuples in an INSERT or UPDATE */
Assert(change->tp.newtuple); Assert(change->data.tp.newtuple);
desc = RelationGetDescr(relation); desc = RelationGetDescr(relation);
@ -2563,8 +2551,9 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
isnull = palloc0(sizeof(bool) * desc->natts); isnull = palloc0(sizeof(bool) * desc->natts);
free = palloc0(sizeof(bool) * desc->natts); free = palloc0(sizeof(bool) * desc->natts);
heap_deform_tuple(&change->tp.newtuple->tuple, desc, newtup = change->data.tp.newtuple;
attrs, isnull);
heap_deform_tuple(&newtup->tuple, desc, attrs, isnull);
for (natt = 0; natt < desc->natts; natt++) for (natt = 0; natt < desc->natts; natt++)
{ {
@ -2628,10 +2617,14 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
dlist_foreach(it, &ent->chunks) dlist_foreach(it, &ent->chunks)
{ {
bool isnull; bool isnull;
ReorderBufferTupleBuf *tup = ReorderBufferChange *cchange;
dlist_container(ReorderBufferChange, node, it.cur)->tp.newtuple; ReorderBufferTupleBuf *ctup;
Pointer chunk = Pointer chunk;
DatumGetPointer(fastgetattr(&tup->tuple, 3, toast_desc, &isnull));
cchange = dlist_container(ReorderBufferChange, node, it.cur);
ctup = cchange->data.tp.newtuple;
chunk = DatumGetPointer(
fastgetattr(&ctup->tuple, 3, toast_desc, &isnull));
Assert(!isnull); Assert(!isnull);
Assert(!VARATT_IS_EXTERNAL(chunk)); Assert(!VARATT_IS_EXTERNAL(chunk));
@ -2665,21 +2658,19 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
* passed to the output plugin. We can't directly heap_fill_tuple() into * passed to the output plugin. We can't directly heap_fill_tuple() into
* the tuplebuf because attrs[] will point back into the current content. * the tuplebuf because attrs[] will point back into the current content.
*/ */
newtup = heap_form_tuple(desc, attrs, isnull); tmphtup = heap_form_tuple(desc, attrs, isnull);
Assert(change->tp.newtuple->tuple.t_len <= MaxHeapTupleSize); Assert(newtup->tuple.t_len <= MaxHeapTupleSize);
Assert(&change->tp.newtuple->header == change->tp.newtuple->tuple.t_data); Assert(&newtup->header == newtup->tuple.t_data);
memcpy(change->tp.newtuple->tuple.t_data, memcpy(newtup->tuple.t_data, tmphtup->t_data, tmphtup->t_len);
newtup->t_data, newtup->tuple.t_len = tmphtup->t_len;
newtup->t_len);
change->tp.newtuple->tuple.t_len = newtup->t_len;
/* /*
* free resources we won't further need, more persistent stuff will be * free resources we won't further need, more persistent stuff will be
* free'd in ReorderBufferToastReset(). * free'd in ReorderBufferToastReset().
*/ */
RelationClose(toast_rel); RelationClose(toast_rel);
pfree(newtup); pfree(tmphtup);
for (natt = 0; natt < desc->natts; natt++) for (natt = 0; natt < desc->natts; natt++)
{ {
if (free[natt]) if (free[natt])

39
src/include/replication/reorderbuffer.h
View File

@ -32,12 +32,22 @@ typedef struct ReorderBufferTupleBuf
char data[MaxHeapTupleSize]; char data[MaxHeapTupleSize];
} ReorderBufferTupleBuf; } ReorderBufferTupleBuf;
/* types of the change passed to a 'change' callback */ /*
* Types of the change passed to a 'change' callback.
*
* For efficiency and simplicity reasons we want to keep Snapshots, CommandIds
* and ComboCids in the same list with the user visible INSERT/UPDATE/DELETE
* changes. Users of the decoding facilities will never see changes with
* *_INTERNAL_* actions.
*/
enum ReorderBufferChangeType enum ReorderBufferChangeType
{ {
REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INSERT,
REORDER_BUFFER_CHANGE_UPDATE, REORDER_BUFFER_CHANGE_UPDATE,
REORDER_BUFFER_CHANGE_DELETE REORDER_BUFFER_CHANGE_DELETE,
REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT,
REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID,
REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID
}; };
/* /*
@ -51,13 +61,8 @@ typedef struct ReorderBufferChange
{ {
XLogRecPtr lsn; XLogRecPtr lsn;
/* type of change */ /* The type of change. */
union enum ReorderBufferChangeType action;
{
enum ReorderBufferChangeType action;
/* do not leak internal enum values to the outside */
int action_internal;
};
/* /*
* Context data for the change, which part of the union is valid depends * Context data for the change, which part of the union is valid depends
@ -65,7 +70,7 @@ typedef struct ReorderBufferChange
*/ */
union union
{ {
/* old, new tuples when action == *_INSERT|UPDATE|DELETE */ /* Old, new tuples when action == *_INSERT|UPDATE|DELETE */
struct struct
{ {
/* relation that has been changed */ /* relation that has been changed */
@ -76,13 +81,19 @@ typedef struct ReorderBufferChange
ReorderBufferTupleBuf *newtuple; ReorderBufferTupleBuf *newtuple;
} tp; } tp;
/* new snapshot */ /* New snapshot, set when action == *_INTERNAL_SNAPSHOT */
Snapshot snapshot; Snapshot snapshot;
/* new command id for existing snapshot in a catalog changing tx */ /*
* New command id for existing snapshot in a catalog changing tx. Set
* when action == *_INTERNAL_COMMAND_ID.
*/
CommandId command_id; CommandId command_id;
/* new cid mapping for catalog changing transaction */ /*
* New cid mapping for catalog changing transaction, set when action
* == *_INTERNAL_TUPLECID.
*/
struct struct
{ {
RelFileNode node; RelFileNode node;
@ -91,7 +102,7 @@ typedef struct ReorderBufferChange
CommandId cmax; CommandId cmax;
CommandId combocid; CommandId combocid;
} tuplecid; } tuplecid;
}; } data;
/* /*
* While in use this is how a change is linked into a transactions, * While in use this is how a change is linked into a transactions,